cyber: fix io hang (#9152)

* cyber: fix io hang

* cyber: code style

* cyber: add io wait for context notify

* cyber: add shutdown check in io example

cyber/io/example/tcp_echo_client.cc

@@ -64,6 +64,9 @@ int main(int argc, char* argv[]) {
           std::cout << "please enter a message (enter Ctrl+C to exit):"
                     << std::endl;
           std::getline(std::cin, user_input);
+          if (!apollo::cyber::OK()) {
+            break;
+          }
           if (user_input.empty()) {
             continue;
           }
@@ -103,4 +106,5 @@ int main(int argc, char* argv[]) {
       "echo_client");
 
   apollo::cyber::WaitForShutdown();
+  return 0;
 }

cyber/io/example/udp_echo_client.cc

@@ -64,6 +64,9 @@ int main(int argc, char* argv[]) {
           std::cout << "please enter a message (enter Ctrl+C to exit):"
                     << std::endl;
           std::getline(std::cin, user_input);
+          if (!apollo::cyber::OK()) {
+            break;
+          }
           if (user_input.empty()) {
             continue;
           }

cyber/io/poll_handler.cc

@@ -17,6 +17,7 @@
 #include "cyber/io/poll_handler.h"
 #include "cyber/common/log.h"
 #include "cyber/io/poller.h"
+#include "cyber/scheduler/scheduler_factory.h"
 
 namespace apollo {
 namespace cyber {
@@ -105,7 +106,7 @@ void PollHandler::ResponseCallback(const PollResponse& rsp) {
   response_ = rsp;
 
   if (routine_->state() == RoutineState::IO_WAIT) {
-    routine_->SetUpdateFlag();
+    scheduler::Instance()->NotifyTask(routine_->id());
   }
 }
 

cyber/scheduler/policy/scheduler_choreography.cc

@@ -258,7 +258,8 @@ bool SchedulerChoreography::NotifyProcessor(uint64_t crid) {
     auto it = id_cr_.find(crid);
     if (it != id_cr_.end()) {
       cr = it->second;
-      if (cr->state() == RoutineState::DATA_WAIT) {
+      if (cr->state() == RoutineState::DATA_WAIT ||
+          cr->state() == RoutineState::IO_WAIT) {
         cr->SetUpdateFlag();
       }
     } else {

cyber/scheduler/policy/scheduler_classic.cc

@@ -173,7 +173,8 @@ bool SchedulerClassic::NotifyProcessor(uint64_t crid) {
     ReadLockGuard<AtomicRWLock> lk(id_cr_lock_);
     if (id_cr_.find(crid) != id_cr_.end()) {
       auto cr = id_cr_[crid];
-      if (cr->state() == RoutineState::DATA_WAIT) {
+      if (cr->state() == RoutineState::DATA_WAIT ||
+          cr->state() == RoutineState::IO_WAIT) {
         cr->SetUpdateFlag();
       }