Android7.0 MessageQueue详解

public static void prepare() {
 prepare(true);
}

private static void prepare(boolean quitallowed) {
 if (sthreadlocal.get() != null) {
 throw new runtimeexception("only one looper may be created per thread");
 }
 //sthreadlocal为线程本地存储区；每个线程仅有一个looper
 sthreadlocal.set(new looper(quitallowed));
}

private looper(boolean quitallowed) {
 //创建出messagequeue
 mqueue = new messagequeue(quitallowed);
 mthread = thread.currentthread();
}

messagequeue(boolean quitallowed) {
 mquitallowed = quitallowed;
 //mptr的类型为long？
 mptr = nativeinit();
}

static jlong android_os_messagequeue_nativeinit(jnienv* env, jclass clazz) {
 //messagequeue的native层实体
 nativemessagequeue* nativemessagequeue = new nativemessagequeue();
 ............
 //这里应该类似与将指针转化成long类型，放在java层保存；估计java层使用时，会在native层将long变成指针，就可以操作队列了
 return reinterpret_cast<jlong>(nativemessagequeue);
}

nativemessagequeue::nativemessagequeue() :
 mpollenv(null), mpollobj(null), mexceptionobj(null) {
 //创建一个native层的looper，也是线程唯一的
 mlooper = looper::getforthread();
 if (mlooper == null) {
 mlooper = new looper(false);
 looper::setforthread(mlooper);
 }
}

looper::looper(bool allownoncallbacks) :
 mallownoncallbacks(allownoncallbacks), msendingmessage(false),
 mpolling(false), mepollfd(-1), mepollrebuildrequired(false),
 mnextrequestseq(0), mresponseindex(0), mnextmessageuptime(llong_max) {
 //此处创建了个fd
 mwakeeventfd = eventfd(0, efd_nonblock | efd_cloexec);
 .......
 rebuildepolllocked();
}

void looper::rebuildepolllocked() {
 // close old epoll instance if we have one.
 if (mepollfd >= 0) {
 close(mepollfd);
 }

 // allocate the new epoll instance and register the wake pipe.
 mepollfd = epoll_create(epoll_size_hint);
 ............
 struct epoll_event eventitem;
 memset(& eventitem, 0, sizeof(epoll_event)); // zero out unused members of data field union
 eventitem.events = epollin;
 eventitem.data.fd = mwakeeventfd;
 //在mepollfd上监听mwakeeventfd上是否有数据到来
 int result = epoll_ctl(mepollfd, epoll_ctl_add, mwakeeventfd, & eventitem);
 ...........
 for (size_t i = 0; i < mrequests.size(); i++) {
 const request& request = mrequests.valueat(i);
 struct epoll_event eventitem;
 request.initeventitem(&eventitem);
 //监听request对应fd上数据的到来
 int epollresult = epoll_ctl(mepollfd, epoll_ctl_add, request.fd, & eventitem);
 ............
 }
}

boolean enqueuemessage(message msg, long when) {
 if (msg.target == null) {
 throw new illegalargumentexception("message must have a target.");
 }
 if (msg.isinuse()) {
 throw new illegalstateexception(msg + " this message is already in use.");
 }

 synchronized (this) {
 if (mquitting) {
  .....
  return false;
 }

 msg.markinuse();
 msg.when = when;
 message p = mmessages;
 boolean needwake;
 if (p == null || when == 0 || when < p.when) {
  // new head, wake up the event queue if blocked.
  msg.next = p;
  mmessages = msg;
  //在头部插入数据，如果之前messagequeue是阻塞的，那么现在需要唤醒
  needwake = mblocked;
 } else {
  // inserted within the middle of the queue. usually we don't have to wake
  // up the event queue unless there is a barrier at the head of the queue
  // and the message is the earliest asynchronous message in the queue.
  needwake = mblocked && p.target == null && msg.isasynchronous();
  message prev;
  for (;;) {
  prev = p;
  p = p.next;
  if (p == null || when < p.when) {
   break;
  }
  //不是第一个异步消息时，needwake置为false
  if (needwake && p.isasynchronous()) {
   needwake = false;
  }
  }
  msg.next = p; // invariant: p == prev.next
  prev.next = msg;
 }
 // we can assume mptr != 0 because mquitting is false.
 if (needwake) {
  nativewake(mptr);
 }
 }
 return true；
}

void nativemessagequeue::wake() {
 mlooper->wake();
}

void looper::wake() {
 uint64_t inc = 1;
 //就是向mwakeeventfd写入数据
 ssize_t nwrite = temp_failure_retry(write(mwakeeventfd, &inc, sizeof(uint64_t)));
 .............
}

void looper::sendmessage(const sp<messagehandler>& handler, const message& message) {
 nsecs_t now = systemtime(system_time_monotonic);
 sendmessageattime(now, handler, message);
}

void looper::sendmessageattime(nsecs_t uptime, const sp<messagehandler>& handler,
 const message& message) {
 size_t i = 0;
 {
 automutex _l(mlock);

 //同样需要按时间插入
 size_t messagecount = mmessageenvelopes.size();
 while (i < messagecount && uptime >= mmessageenvelopes.itemat(i).uptime) {
  i += 1;
 }

 //将message包装成一个messageenvelope对象
 messageenvelope messageenvelope(uptime, handler, message);
 mmessageenvelopes.insertat(messageenvelope, i, 1);

 // optimization: if the looper is currently sending a message, then we can skip
 // the call to wake() because the next thing the looper will do after processing
 // messages is to decide when the next wakeup time should be. in fact, it does
 // not even matter whether this code is running on the looper thread.
 if (msendingmessage) {
  return;
 }
 }
 // wake the poll loop only when we enqueue a new message at the head.
 if (i == 0) {
 //若插入在队列头部，同样利用wake函数触发epoll唤醒
 wake();
 }
}

public static void loop() {
 final looper me = mylooper();
 .......
 for (;;) {
 message msg = queue.next(); // might block
 .......
 try {
  //调用message的处理函数进行处理
  msg.target.dispatchmessage(msg);
 }........
 }
}

message next() {
 //mptr保存了nativemessagequeue的指针
 final long ptr = mptr;
 .......
 int pendingidlehandlercount = -1; // -1 only during first iteration
 int nextpolltimeoutmillis = 0;

 for (;;) {
 if (nextpolltimeoutmillis != 0) {
  //会调用native函数，最终调用ipcthread的talkwithdriver，将数据写入binder驱动或者读取一次数据
  //不知道在此处进行这个操作的理由？
  binder.flushpendingcommands();
 }

 //处理native层的数据，此处会利用epoll进行blocked
 nativepollonce(ptr, nextpolltimeoutmillis);

 synchronized (this) {
  final long now = systemclock.uptimemillis();
  message prevmsg = null;
  message msg = mmessages;
  //下面其实就是找出下一个异步处理类型的消息；异步处理类型的消息，才含有对应的执行函数
  if (msg != null && msg.target == null) {
  // stalled by a barrier. find the next asynchronous message in the queue.
  do {
   prevmsg = msg;
   msg = msg.next;
  } while (msg != null && !msg.isasynchronous());
  }

  if (msg != null) {
  if (now < msg.when) {
   // next message is not ready. set a timeout to wake up when it is ready.
   nextpolltimeoutmillis = (int) math.min(msg.when - now, integer.max_value);
  } else {
   // got a message.
   mblocked = false;
   //完成next记录的存储
   if (prevmsg != null) {
   prevmsg.next = msg.next;
   } else {
   mmessages = msg.next;
   }
   msg.next = null;
   if (debug) log.v(tag, "returning message: " + msg);
   msg.markinuse();
   return msg;
  }
  } else {
  // no more messages.
  nextpolltimeoutmillis = -1;
  }

  // process the quit message now that all pending messages have been handled.
  if (mquitting) {
  dispose();
  return null;
  }

  //messagequeue中引入了idlehandler接口，即当messagequeue没有数据处理时，调用idlehandler进行一些工作

  //pendingidlehandlercount表示待处理的idlehandler，初始为-1
  if (pendingidlehandlercount < 0
   && (mmessages == null || now < mmessages.when)) {
  //midlehandlers的size默认为0，调用接口addidlehandler才能增加
  pendingidlehandlercount = midlehandlers.size();
  }

  if (pendingidlehandlercount <= 0) {
  // no idle handlers to run. loop and wait some more.
  mblocked = true;
  continue;
  }

  //将待处理的idlehandler加入到pendingidlehandlers中
  if (mpendingidlehandlers == null) {
  mpendingidlehandlers = new idlehandler[math.max(pendingidlehandlercount, 4)];
  }
  //调用arraylist.toarray(t[])节省每次分配的开销；毕竟对于message.next这样调用频率较高的函数，能省一点就是一点
  mpendingidlehandlers = midlehandlers.toarray(mpendingidlehandlers);
 }

 for (int i = 0; i < pendingidlehandlercount; i++) {
  final idlehandler idler = mpendingidlehandlers[i];
  mpendingidlehandlers[i] = null; // release the reference to the handler

  boolean keep = false;
  try {
  //执行实现类的queueidle函数，返回值决定是否继续保留
  keep = idler.queueidle();
  } catch (throwable t) {
  log.wtf(tag, "idlehandler threw exception", t);
  }

  if (!keep) {
  synchronized (this) {
   midlehandlers.remove(idler);
  }
  }
 }
 pendingidlehandlercount = 0;
 nextpolltimeoutmillis = 0;
 }
}

static void android_os_messagequeue_nativepollonce(jnienv* env, jobject obj,
 jlong ptr, jint timeoutmillis) {
 //果然java层调用native层messagequeue时，将long类型的ptr变为指针
 nativemessagequeue* nativemessagequeue = reinterpret_cast<nativemessagequeue*>(ptr);
 nativemessagequeue->pollonce(env, obj, timeoutmillis);
}

void nativemessagequeue::pollonce(jnienv* env, jobject pollobj, int timeoutmillis) {
 mpollenv = env;
 mpollobj = pollobj;
 //最后还是进入到native层looper的pollonce函数
 mlooper->pollonce(timeoutmillis);
 mpollobj = null;
 mpollenv = null;

 if (mexceptionobj) {
 .........
 }
}

//timeoutmillis为超时等待时间。值为-1时，表示无限等待直到有事件到来；值为0时，表示无需等待
//outfd此时为null，含义是：存储产生事件的文件句柄
//outevents此时为null，含义是：存储outfd上发生了哪些事件，包括可读、可写、错误和中断
//outdata此时为null，含义是：存储上下文数据，其实调用时传入的参数
int looper::pollonce(int timeoutmillis, int* outfd, int* outevents, void** outdata) {
 int result = 0;
 for (;;) {
 //处理response，目前我们先不关注response的内含
 while (mresponseindex < mresponses.size()) {
  const response& response = mresponses.itemat(mresponseindex++);
  int ident = response.request.ident;
  if (ident >= 0) {
  int fd = response.request.fd;
  int events = response.events;
  void* data = response.request.data;

  if (outfd != null) *outfd = fd;
  if (outevents != null) *outevents = events;
  if (outdata != null) *outdata = data;
  return ident;
  }
 }

 //根据pollinner的结果，进行操作
 if (result != 0) {
  if (outfd != null) *outfd = 0;
  if (outevents != null) *outevents = 0;
  if (outdata != null) *outdata = null;
  return result;
 }

 //主力还是靠pollinner
 result = pollinner(timeoutmillis);
 }
}

int looper::pollinner(int timeoutmillis) {
 // adjust the timeout based on when the next message is due.
 //timeoutmillis是java层事件等待事件
 //native层维持了native message的等待时间
 //此处其实就是选择最小的等待时间
 if (timeoutmillis != 0 && mnextmessageuptime != llong_max) {
  nsecs_t now = systemtime(system_time_monotonic);
  int messagetimeoutmillis = tomillisecondtimeoutdelay(now, mnextmessageuptime);
  if (messagetimeoutmillis >= 0
  && (timeoutmillis < 0 || messagetimeoutmillis < timeoutmillis)) {
  timeoutmillis = messagetimeoutmillis;
 }
 }

 int result = poll_wake;
 //pollinner初始就清空response
 mresponses.clear();
 mresponseindex = 0;

 // we are about to idle.
 mpolling = true;

 //利用epoll等待mepollfd监控的句柄上事件到达
 struct epoll_event eventitems[epoll_max_events];
 int eventcount = epoll_wait(mepollfd, eventitems, epoll_max_events, timeoutmillis);

 // no longer idling.
 mpolling = false;

 // acquire lock.
 mlock.lock();

 //重新调用rebuildepolllocked时，将使得epoll句柄能够监听新加入request对应的fd
 if (mepollrebuildrequired) {
 mepollrebuildrequired = false;
 rebuildepolllocked();
 goto done;
 }

 // check for poll error.
 if (eventcount < 0) {
 if (errno == eintr) {
  goto done;
 }
 ......
 result = poll_error;
 goto done;
 }

 // check for poll timeout.
 if (eventcount == 0) {
 result = poll_timeout;
 goto done;
 }

 for (int i = 0; i < eventcount; i++) {
 if (fd == mwakeeventfd) {
  if (epollevents & epollin) {
  //前面已经分析过，当java层或native层有数据写入队列时，将写mwakeeventfd，以触发epoll唤醒
  //awoken将读取并清空mwakeeventfd上的数据
  awoken();
  } else {
  .........
  }
 } else {
  //epoll同样监听的request对应的fd
  ssize_t requestindex = mrequests.indexofkey(fd);
  if (requestindex >= 0) {
  int events = 0;
  if (epollevents & epollin) events |= event_input;
  if (epollevents & epollout) events |= event_output;
  if (epollevents & epollerr) events |= event_error;
  if (epollevents & epollhup) events |= event_hangup;
  //存储这个fd对应的response
  pushresponse(events, mrequests.valueat(requestindex));
  } else {
  ..........
  }
 }
 }

 // invoke pending message callbacks.
 mnextmessageuptime = llong_max;
 //处理native层的message
 while (mmessageenvelopes.size() != 0) {
 nsecs_t now = systemtime(system_time_monotonic);
 const messageenvelope& messageenvelope = mmessageenvelopes.itemat(0);
 if (messageenvelope.uptime <= now) {
  // remove the envelope from the list.
  // we keep a strong reference to the handler until the call to handlemessage
  // finishes. then we drop it so that the handler can be deleted *before*
  // we reacquire our lock.
  {
  sp<messagehandler> handler = messageenvelope.handler;
  message message = messageenvelope.message;
  mmessageenvelopes.removeat(0);
  msendingmessage = true;
  mlock.unlock();

  //处理native message
  handler->handlemessage(message);
  }
  mlock.lock();
  msendingmessage = false;
  result = poll_callback;
 } else {
  // the last message left at the head of the queue determines the next wakeup time.
  mnextmessageuptime = messageenvelope.uptime;
  break;
 }
 }

 // release lock.
 mlock.unlock();

 //处理带回调函数的response
 for (size_t i = 0; i < mresponses.size(); i++) {
 response& response = mresponses.edititemat(i);
 if (response.request.ident == poll_callback) {
  int fd = response.request.fd;
  int events = response.events;
  void* data = response.request.data;

  //调用response的callback
  int callbackresult = response.request.callback->handleevent(fd, events, data);
  if (callbackresult == 0) {
  removefd(fd, response.request.seq);
  }

  response.request.callback.clear();
  result = poll_callback;
 }
 }
 return result;
}

//fd表示需要监听的句柄
//ident的含义还没有搞明白
//events表示需要监听的事件，例如event_input、event_output、event_error和event_hangup中的一个或多个
//callback为事件发生后的回调函数
//data为回调函数对应的参数
int looper::addfd(int fd, int ident, int events, looper_callbackfunc callback, void* data) {
 return addfd(fd, ident, events, callback ? new simpleloopercallback(callback) : null, data);
}

int looper::addfd(int fd, int ident, int events, const sp<loopercallback>& callback, void* data) {
 ........
 {
 automutex _l(mlock);

 //利用参数构造一个request
 request request;
 request.fd = fd;
 request.ident = ident;
 request.events = events;
 request.seq = mnextrequestseq++;
 request.callback = callback;
 request.data = data;
 if (mnextrequestseq == -1) mnextrequestseq = 0; // reserve sequence number -1

 struct epoll_event eventitem;
 request.initeventitem(&eventitem);

 //判断之前是否已经利用该fd构造过request
 ssize_t requestindex = mrequests.indexofkey(fd);
 if (requestindex < 0) {
  //mepollfd新增一个需监听fd
  int epollresult = epoll_ctl(mepollfd, epoll_ctl_add, fd, & eventitem);
  .......
  mrequests.add(fd, request);
 } else {
  //mepollfd修改旧的fd对应的监听事件
  int epollresult = epoll_ctl(mepollfd, epoll_ctl_mod, fd, & eventitem);
  if (epollresult < 0) {
  if (errno == enoent) {
   // tolerate enoent because it means that an older file descriptor was
   // closed before its callback was unregistered and meanwhile a new
   // file descriptor with the same number has been created and is now
   // being registered for the first time. 
   epollresult = epoll_ctl(mepollfd, epoll_ctl_add, fd, & eventitem);
   .......
  }
  //发生错误重新加入时，安排epollrebuildlocked，将让epollfd重新添加一次待监听的fd
  scheduleepollrebuildlocked();
  }
  mrequests.replacevalueat(requestindex, request);
 }
 }
}