| 1088 | } |
| 1089 | |
| 1090 | void TNonblockingServer::registerEvents(event_base* user_event_base) { |
| 1091 | userEventBase_ = user_event_base; |
| 1092 | |
| 1093 | // init listen socket |
| 1094 | if (serverSocket_ == THRIFT_INVALID_SOCKET) |
| 1095 | createAndListenOnSocket(); |
| 1096 | |
| 1097 | // set up the IO threads |
| 1098 | assert(ioThreads_.empty()); |
| 1099 | if (!numIOThreads_) { |
| 1100 | numIOThreads_ = DEFAULT_IO_THREADS; |
| 1101 | } |
| 1102 | // User-provided event-base doesn't works for multi-threaded servers |
| 1103 | assert(numIOThreads_ == 1 || !userEventBase_); |
| 1104 | |
| 1105 | for (uint32_t id = 0; id < numIOThreads_; ++id) { |
| 1106 | // the first IO thread also does the listening on server socket |
| 1107 | THRIFT_SOCKET listenFd = (id == 0 ? serverSocket_ : THRIFT_INVALID_SOCKET); |
| 1108 | |
| 1109 | shared_ptr<TNonblockingIOThread> thread( |
| 1110 | new TNonblockingIOThread(this, id, listenFd, useHighPriorityIOThreads_)); |
| 1111 | ioThreads_.push_back(thread); |
| 1112 | } |
| 1113 | |
| 1114 | // Notify handler of the preServe event |
| 1115 | if (eventHandler_) { |
| 1116 | eventHandler_->preServe(); |
| 1117 | } |
| 1118 | |
| 1119 | // Start all of our helper IO threads. Note that the threads run forever, |
| 1120 | // only terminating if stop() is called. |
| 1121 | assert(ioThreads_.size() == numIOThreads_); |
| 1122 | assert(ioThreads_.size() > 0); |
| 1123 | |
| 1124 | TOutput::instance().printf("TNonblockingServer: Serving with %d io threads.", |
| 1125 | ioThreads_.size()); |
| 1126 | |
| 1127 | // Launch all the secondary IO threads in separate threads |
| 1128 | if (ioThreads_.size() > 1) { |
| 1129 | ioThreadFactory_.reset(new ThreadFactory( |
| 1130 | false // detached |
| 1131 | )); |
| 1132 | |
| 1133 | assert(ioThreadFactory_.get()); |
| 1134 | |
| 1135 | // intentionally starting at thread 1, not 0 |
| 1136 | for (uint32_t i = 1; i < ioThreads_.size(); ++i) { |
| 1137 | shared_ptr<Thread> thread = ioThreadFactory_->newThread(ioThreads_[i]); |
| 1138 | ioThreads_[i]->setThread(thread); |
| 1139 | thread->start(); |
| 1140 | } |
| 1141 | } |
| 1142 | |
| 1143 | // Register the events for the primary (listener) IO thread |
| 1144 | ioThreads_[0]->registerEvents(); |
| 1145 | } |
| 1146 | |
| 1147 | /** |