fun run() { while (doWork) { deliverLock() while (tomLayer.isRetrievingState()) { println("-- Retrieving State") canDeliver.awaitUninterruptibly() if (tomLayer.getLastExec() === -1) println("-- Ready to process operations") } try { val decisions: ArrayList<Decision> = ArrayList<Decision>() decidedLock.lock() if (decided.isEmpty()) { notEmptyQueue.await() } decided.drainTo(decisions) decidedLock.unlock() if (!doWork) break if (decisions.size() > 0) { val requests: Array<Array<TOMMessage>> = arrayOfNulls<Array<TOMMessage>>(decisions.size()) val consensusIds = IntArray(requests.size) val leadersIds = IntArray(requests.size) val regenciesIds = IntArray(requests.size) var cDecs: Array<CertifiedDecision?> cDecs = arrayOfNulls<CertifiedDecision>(requests.size) var count = 0 for (d in decisions) { requests[count] = extractMessagesFromDecision(d) consensusIds[count] = d.getConsensusId() leadersIds[count] = d.getLeader() regenciesIds[count] = d.getRegency() val cDec = CertifiedDecision( this.controller.getStaticConf().getProcessId(), d.getConsensusId(), d.getValue(), d.getDecisionEpoch().proo ) cDecs[count] = cDec if (requests[count][0].equals(d.firstMessageProposed)) { val time: Long = requests[count][0].timestamp val seed: Long = requests[count][0].seed val numOfNonces: Int = requests[count][0].numOfNonces requests[count][0] = d.firstMessageProposed requests[count][0].timestamp = time requests[count][0].seed = seed requests[count][0].numOfNonces = numOfNonces } count++ } val lastDecision: Decision = decisions[decisions.size() - 1] if (requests != null && requests.size > 0) { deliverMessages(consensusIds, regenciesIds, leadersIds, cDecs, requests) if (controller.hasUpdates()) { processReconfigMessages(lastDecision.getConsensusId()) tomLayer.setLastExec(lastDecision.getConsensusId()) tomLayer.setInExec(-1) } } val cid: Int = lastDecision.getConsensusId() if (cid > 2) { val stableConsensus = cid – 3 tomLayer.execManager.removeConsensus(stableConsensus) } } } catch (e: Exception) { e.printStackTrace(System.err) } deliverUnlock() } Logger.getLogger(DeliveryThread::class.java.getName()).log(Level.INFO, "DeliveryThread stopped.") }
This is the code for the thread . It delivers decisions to the TOM request receiver object ( which is the application )