Showing the changes to assignment-client/src/AssignmentClient.cpp (1 of 107 files). Show all
Unified · Split
| @@ -62,6 +62,17 @@ AssignmentClient::AssignmentClient(Assignment::Type requestAssignmentType, QStri · 61 unmodified lines | ||
| 62 | 62 | DependencyManager::registerInheritance<EntityActionFactoryInterface, AssignmentActionFactory>(); |
| 63 | 63 | auto actionFactory = DependencyManager::set<AssignmentActionFactory>(); |
| 64 | 64 | |
| 65 | + // setup a thread for the NodeList and its PacketReceiver | |
| 66 | + QThread* nodeThread = new QThread(this); | |
| 67 | + nodeThread->setObjectName("NodeList Thread"); | |
| 68 | + nodeThread->start(); | |
| 69 | + | |
| 70 | + // make sure the node thread is given highest priority | |
| 71 | + nodeThread->setPriority(QThread::TimeCriticalPriority); | |
| 72 | + | |
| 73 | + // put the NodeList on the node thread | |
| 74 | + nodeList->moveToThread(nodeThread); | |
| 75 | + | |
| 65 | 76 | // make up a uuid for this child so the parent can tell us apart. This id will be changed |
| 66 | 77 | // when the domain server hands over an assignment. |
| 67 | 78 | QUuid nodeUUID = QUuid::createUuid(); |
| @@ -104,9 +115,6 @@ AssignmentClient::AssignmentClient(Assignment::Type requestAssignmentType, QStri · 36 unmodified lines | ||
| 104 | 115 | connect(&_requestTimer, SIGNAL(timeout()), SLOT(sendAssignmentRequest())); |
| 105 | 116 | _requestTimer.start(ASSIGNMENT_REQUEST_INTERVAL_MSECS); |
| 106 | 117 | |
| 107 | − // connect our readPendingDatagrams method to the readyRead() signal of the socket | |
| 108 | − connect(&nodeList->getNodeSocket(), &QUdpSocket::readyRead, this, &AssignmentClient::readPendingDatagrams); | |
| 109 | − | |
| 110 | 118 | // connections to AccountManager for authentication |
| 111 | 119 | connect(&AccountManager::getInstance(), &AccountManager::authRequired, |
| 112 | 120 | this, &AssignmentClient::handleAuthenticationRequest); |
| @@ -121,11 +129,12 @@ AssignmentClient::AssignmentClient(Assignment::Type requestAssignmentType, QStri · 8 unmodified lines | ||
| 121 | 129 | qDebug() << "Assignment-client monitor socket is" << _assignmentClientMonitorSocket; |
| 122 | 130 | |
| 123 | 131 | // Hook up a timer to send this child's status to the Monitor once per second |
| 124 | − setUpStatsToMonitor(); | |
| 132 | + setUpStatusToMonitor(); | |
| 125 | 133 | } |
| 134 | + auto& packetReceiver = DependencyManager::get<NodeList>()->getPacketReceiver(); | |
| 135 | + packetReceiver.registerListener(PacketType::CreateAssignment, this, "handleCreateAssignmentPacket"); | |
| 136 | + packetReceiver.registerListener(PacketType::StopNode, this, "handleStopNodePacket"); | |
| 126 | 137 | } |
| 127 | − | |
| 128 | − | |
| 129 | 138 | void AssignmentClient::stopAssignmentClient() { |
| 130 | 139 | qDebug() << "Forced stop of assignment-client."; |
| 131 | 140 | |
| @@ -151,6 +160,16 @@ void AssignmentClient::stopAssignmentClient() { · 19 unmodified lines | ||
| 151 | 160 | } |
| 152 | 161 | } |
| 153 | 162 | |
| 163 | +AssignmentClient::~AssignmentClient() { | |
| 164 | + QThread* nodeThread = DependencyManager::get<NodeList>()->thread(); | |
| 165 | + | |
| 166 | + // remove the NodeList from the DependencyManager | |
| 167 | + DependencyManager::destroy<NodeList>(); | |
| 168 | + | |
| 169 | + // ask the node thread to quit and wait until it is done | |
| 170 | + nodeThread->quit(); | |
| 171 | + nodeThread->wait(); | |
| 172 | +} | |
| 154 | 173 | |
| 155 | 174 | void AssignmentClient::aboutToQuit() { |
| 156 | 175 | stopAssignmentClient(); |
| @@ -160,23 +179,28 @@ void AssignmentClient::aboutToQuit() { · 3 unmodified lines | ||
| 160 | 179 | } |
| 161 | 180 | |
| 162 | 181 | |
| 163 | −void AssignmentClient::setUpStatsToMonitor() { | |
| 182 | +void AssignmentClient::setUpStatusToMonitor() { | |
| 164 | 183 | // send a stats packet every 1 seconds |
| 165 | − connect(&_statsTimerACM, &QTimer::timeout, this, &AssignmentClient::sendStatsPacketToACM); | |
| 184 | + connect(&_statsTimerACM, &QTimer::timeout, this, &AssignmentClient::sendStatusPacketToACM); | |
| 166 | 185 | _statsTimerACM.start(1000); |
| 167 | 186 | } |
| 168 | 187 | |
| 169 | −void AssignmentClient::sendStatsPacketToACM() { | |
| 188 | +void AssignmentClient::sendStatusPacketToACM() { | |
| 170 | 189 | // tell the assignment client monitor what this assignment client is doing (if anything) |
| 171 | − QJsonObject statsObject; | |
| 172 | 190 | auto nodeList = DependencyManager::get<NodeList>(); |
| 191 | + | |
| 192 | + quint8 assignmentType = Assignment::Type::AllTypes; | |
| 173 | 193 | |
| 174 | 194 | if (_currentAssignment) { |
| 175 | − statsObject["assignment_type"] = _currentAssignment->getTypeName(); | |
| 176 | − } else { | |
| 177 | − statsObject["assignment_type"] = "none"; | |
| 195 | + assignmentType = _currentAssignment->getType(); | |
| 178 | 196 | } |
| 179 | − nodeList->sendStats(statsObject, _assignmentClientMonitorSocket); | |
| 197 | + | |
| 198 | + auto statusPacket = NLPacket::create(PacketType::AssignmentClientStatus, sizeof(assignmentType) + NUM_BYTES_RFC4122_UUID); | |
| 199 | + | |
| 200 | + statusPacket->write(nodeList->getSessionUUID().toRfc4122()); | |
| 201 | + statusPacket->writePrimitive(assignmentType); | |
| 202 | + | |
| 203 | + nodeList->sendPacket(std::move(statusPacket), _assignmentClientMonitorSocket); | |
| 180 | 204 | } |
| 181 | 205 | |
| 182 | 206 | void AssignmentClient::sendAssignmentRequest() { |
| @@ -206,85 +230,64 @@ void AssignmentClient::sendAssignmentRequest() { · 23 unmodified lines | ||
| 206 | 230 | } |
| 207 | 231 | } |
| 208 | 232 | |
| 209 | −void AssignmentClient::readPendingDatagrams() { | |
| 210 | − auto nodeList = DependencyManager::get<NodeList>(); | |
| 211 | − | |
| 212 | − QByteArray receivedPacket; | |
| 213 | − HifiSockAddr senderSockAddr; | |
| 233 | +void AssignmentClient::handleCreateAssignmentPacket(QSharedPointer<NLPacket> packet) { | |
| 234 | + qDebug() << "Received a PacketType::CreateAssignment - attempting to unpack."; | |
| 214 | 235 | |
| 215 | − while (nodeList->getNodeSocket().hasPendingDatagrams()) { | |
| 216 | − receivedPacket.resize(nodeList->getNodeSocket().pendingDatagramSize()); | |
| 217 | − nodeList->getNodeSocket().readDatagram(receivedPacket.data(), receivedPacket.size(), | |
| 218 | − senderSockAddr.getAddressPointer(), senderSockAddr.getPortPointer()); | |
| 236 | + // construct the deployed assignment from the packet data | |
| 237 | + _currentAssignment = AssignmentFactory::unpackAssignment(*packet); | |
| 219 | 238 | |
| 220 | − if (nodeList->packetVersionAndHashMatch(receivedPacket)) { | |
| 221 | − if (packetTypeForPacket(receivedPacket) == PacketType::CreateAssignment) { | |
| 222 | − | |
| 223 | − qDebug() << "Received a PacketType::CreateAssignment - attempting to unpack."; | |
| 224 | − | |
| 225 | − // construct the deployed assignment from the packet data | |
| 226 | − _currentAssignment = AssignmentFactory::unpackAssignment(receivedPacket); | |
| 239 | + if (_currentAssignment) { | |
| 240 | + qDebug() << "Received an assignment -" << *_currentAssignment; | |
| 227 | 241 | |
| 228 | − if (_currentAssignment) { | |
| 229 | − qDebug() << "Received an assignment -" << *_currentAssignment; | |
| 242 | + auto nodeList = DependencyManager::get<NodeList>(); | |
| 230 | 243 | |
| 231 | − // switch our DomainHandler hostname and port to whoever sent us the assignment | |
| 244 | + // switch our DomainHandler hostname and port to whoever sent us the assignment | |
| 232 | 245 | |
| 233 | − nodeList->getDomainHandler().setSockAddr(senderSockAddr, _assignmentServerHostname); | |
| 234 | − nodeList->getDomainHandler().setAssignmentUUID(_currentAssignment->getUUID()); | |
| 246 | + nodeList->getDomainHandler().setSockAddr(packet->getSenderSockAddr(), _assignmentServerHostname); | |
| 247 | + nodeList->getDomainHandler().setAssignmentUUID(_currentAssignment->getUUID()); | |
| 235 | 248 | |
| 236 | − qDebug() << "Destination IP for assignment is" << nodeList->getDomainHandler().getIP().toString(); | |
| 249 | + qDebug() << "Destination IP for assignment is" << nodeList->getDomainHandler().getIP().toString(); | |
| 237 | 250 | |
| 238 | − // start the deployed assignment | |
| 239 | − QThread* workerThread = new QThread; | |
| 240 | − workerThread->setObjectName("ThreadedAssignment Worker"); | |
| 251 | + // start the deployed assignment | |
| 252 | + QThread* workerThread = new QThread; | |
| 253 | + workerThread->setObjectName("ThreadedAssignment Worker"); | |
| 241 | 254 | |
| 242 | − connect(workerThread, &QThread::started, _currentAssignment.data(), &ThreadedAssignment::run); | |
| 255 | + connect(workerThread, &QThread::started, _currentAssignment.data(), &ThreadedAssignment::run); | |
| 243 | 256 | |
| 244 | − // Once the ThreadedAssignment says it is finished - we ask it to deleteLater | |
| 245 | − // This is a queued connection so that it is put into the event loop to be processed by the worker | |
| 246 | − // thread when it is ready. | |
| 247 | − connect(_currentAssignment.data(), &ThreadedAssignment::finished, _currentAssignment.data(), | |
| 248 | − &ThreadedAssignment::deleteLater, Qt::QueuedConnection); | |
| 257 | + // Once the ThreadedAssignment says it is finished - we ask it to deleteLater | |
| 258 | + // This is a queued connection so that it is put into the event loop to be processed by the worker | |
| 259 | + // thread when it is ready. | |
| 260 | + connect(_currentAssignment.data(), &ThreadedAssignment::finished, _currentAssignment.data(), | |
| 261 | + &ThreadedAssignment::deleteLater, Qt::QueuedConnection); | |
| 249 | 262 | |
| 250 | − // once it is deleted, we quit the worker thread | |
| 251 | − connect(_currentAssignment.data(), &ThreadedAssignment::destroyed, workerThread, &QThread::quit); | |
| 263 | + // once it is deleted, we quit the worker thread | |
| 264 | + connect(_currentAssignment.data(), &ThreadedAssignment::destroyed, workerThread, &QThread::quit); | |
| 252 | 265 | |
| 253 | − // have the worker thread remove itself once it is done | |
| 254 | − connect(workerThread, &QThread::finished, workerThread, &QThread::deleteLater); | |
| 266 | + // have the worker thread remove itself once it is done | |
| 267 | + connect(workerThread, &QThread::finished, workerThread, &QThread::deleteLater); | |
| 255 | 268 | |
| 256 | − // once the worker thread says it is done, we consider the assignment completed | |
| 257 | − connect(workerThread, &QThread::destroyed, this, &AssignmentClient::assignmentCompleted); | |
| 269 | + // once the worker thread says it is done, we consider the assignment completed | |
| 270 | + connect(workerThread, &QThread::destroyed, this, &AssignmentClient::assignmentCompleted); | |
| 258 | 271 | |
| 259 | − _currentAssignment->moveToThread(workerThread); | |
| 272 | + _currentAssignment->moveToThread(workerThread); | |
| 260 | 273 | |
| 261 | − // move the NodeList to the thread used for the _current assignment | |
| 262 | − nodeList->moveToThread(workerThread); | |
| 274 | + // Starts an event loop, and emits workerThread->started() | |
| 275 | + workerThread->start(); | |
| 276 | + } else { | |
| 277 | + qDebug() << "Received an assignment that could not be unpacked. Re-requesting."; | |
| 278 | + } | |
| 279 | +} | |
| 263 | 280 | |
| 264 | − // let the assignment handle the incoming datagrams for its duration | |
| 265 | − disconnect(&nodeList->getNodeSocket(), 0, this, 0); | |
| 266 | − connect(&nodeList->getNodeSocket(), &QUdpSocket::readyRead, _currentAssignment.data(), | |
| 267 | − &ThreadedAssignment::readPendingDatagrams); | |
| 281 | +void AssignmentClient::handleStopNodePacket(QSharedPointer<NLPacket> packet) { | |
| 282 | + const HifiSockAddr& senderSockAddr = packet->getSenderSockAddr(); | |
| 283 | + | |
| 284 | + if (senderSockAddr.getAddress() == QHostAddress::LocalHost || | |
| 285 | + senderSockAddr.getAddress() == QHostAddress::LocalHostIPv6) { | |
| 286 | + qDebug() << "AssignmentClientMonitor at" << senderSockAddr << "requested stop via PacketType::StopNode."; | |
| 268 | 287 | |
| 269 | − // Starts an event loop, and emits workerThread->started() | |
| 270 | − workerThread->start(); | |
| 271 | − } else { | |
| 272 | − qDebug() << "Received an assignment that could not be unpacked. Re-requesting."; | |
| 273 | − } | |
| 274 | − } else if (packetTypeForPacket(receivedPacket) == PacketType::StopNode) { | |
| 275 | − if (senderSockAddr.getAddress() == QHostAddress::LocalHost || | |
| 276 | − senderSockAddr.getAddress() == QHostAddress::LocalHostIPv6) { | |
| 277 | − qDebug() << "AssignmentClientMonitor at" << senderSockAddr << "requested stop via PacketType::StopNode."; | |
| 278 | − | |
| 279 | − QCoreApplication::quit(); | |
| 280 | − } else { | |
| 281 | − qDebug() << "Got a stop packet from other than localhost."; | |
| 282 | − } | |
| 283 | − } else { | |
| 284 | − // have the NodeList attempt to handle it | |
| 285 | − nodeList->processNodeData(senderSockAddr, receivedPacket); | |
| 286 | − } | |
| 287 | − } | |
| 288 | + QCoreApplication::quit(); | |
| 289 | + } else { | |
| 290 | + qDebug() << "Got a stop packet from other than localhost."; | |
| 288 | 291 | } |
| 289 | 292 | } |
| 290 | 293 | |
| @@ -327,8 +330,8 @@ void AssignmentClient::assignmentCompleted() { · 36 unmodified lines | ||
| 327 | 330 | |
| 328 | 331 | auto nodeList = DependencyManager::get<NodeList>(); |
| 329 | 332 | |
| 330 | − // have us handle incoming NodeList datagrams again, and make sure our ThreadedAssignment isn't handling them | |
| 331 | − connect(&nodeList->getNodeSocket(), &QUdpSocket::readyRead, this, &AssignmentClient::readPendingDatagrams); | |
| 333 | + // tell the packet receiver to stop dropping packets | |
| 334 | + nodeList->getPacketReceiver().setShouldDropPackets(false); | |
| 332 | 335 | |
| 333 | 336 | // reset our NodeList by switching back to unassigned and clearing the list |
| 334 | 337 | nodeList->setOwnerType(NodeType::Unassigned); |