Merge pull request #40 from Chia-Network/timelord-fixes

Timelord fixes

Author: hoffmang9

lib/chiavdf/fast_vdf/vdf_server.cpp

@@ -3,6 +3,7 @@
 using boost::asio::ip::tcp;
 
 const int max_length = 2048;
+const int kMaxProcessesAllowed = 3;
 std::mutex socket_mutex;
 
 void PrintInfo(std::string input) {
@@ -24,22 +25,6 @@ void CreateAndWriteProof(integer D, form x, int64_t num_iterations, WesolowskiCa
     boost::asio::write(sock, boost::asio::buffer(str_result.c_str(), str_result.size()));
 }
 
-
-void PollTimelord(tcp::socket& sock, bool& got_iters) {
-    // Wait for 15s, if no iters come, poll each 5 seconds the timelord.
-    int seconds = 0;
-    while (!got_iters) {
-        std::this_thread::sleep_for (std::chrono::seconds(1));
-        seconds++;
-        if (seconds >= 15 && (seconds - 15) % 5 == 0) {
-            socket_mutex.lock();
-            boost::asio::write(sock, boost::asio::buffer("POLL", 4));
-            socket_mutex.unlock();
-        }
-    }
-}
-
-
 void session(tcp::socket sock) {
     try {
         char disc[350];
@@ -107,7 +92,6 @@ void session(tcp::socket sock) {
         bool stopped = false;
         bool got_iters = false;
         std::thread vdf_worker(repeated_square, f, D, L, std::ref(weso), std::ref(stopped));
-        std::thread poll_thread(PollTimelord, std::ref(sock), std::ref(got_iters));
 
         // Tell client that I'm ready to get the challenges.
         boost::asio::write(sock, boost::asio::buffer("OK", 2));
@@ -125,7 +109,6 @@ void session(tcp::socket sock) {
             if (iters == 0) {
                 PrintInfo("Got stop signal!");
                 stopped = true;
-                poll_thread.join();
                 for (int i = 0; i < threads.size(); i++)
                     stop_vector[i] = true;
                 for (int t = 0; t < threads.size(); t++) {
@@ -135,7 +118,7 @@ void session(tcp::socket sock) {
                 free(forms);
             } else {
                 int max_iter = 0;
-                int max_iter_thread_id;
+                int max_iter_thread_id = -1;
                 int min_iter = std::numeric_limits<int> :: max();
                 bool unique = true;
                 for (auto active_iter: seen_iterations) {
@@ -155,13 +138,13 @@ void session(tcp::socket sock) {
                     PrintInfo("Duplicate iteration " + to_string(iters) + "... Ignoring.");
                     continue;
                 }
-                if (threads.size() < 3 || iters < min_iter) {
+                if (threads.size() < kMaxProcessesAllowed || iters < min_iter) {
                     seen_iterations.insert({iters, threads.size()});
                     PrintInfo("Running proving for iter: " + to_string(iters));
                     stop_vector[threads.size()] = false;
                     threads.push_back(std::thread(CreateAndWriteProof, D, f, iters, std::ref(weso), 
                                       std::ref(stop_vector[threads.size()]), std::ref(sock)));
-                    if (threads.size() > 3) {
+                    if (threads.size() > kMaxProcessesAllowed) {
                         PrintInfo("Stopping proving for iter: " + to_string(max_iter));
                         stop_vector[max_iter_thread_id] = true;
                         seen_iterations.erase({max_iter, max_iter_thread_id});

src/timelord.py

@@ -149,7 +149,7 @@ async def _update_avg_ips(self, challenge_hash, iterations_needed, ip):
                     new_avg_ips = int((prev_avg_ips * trials + ips) / (trials + 1))
                     self.avg_ips[ip] = (new_avg_ips, trials + 1)
                     log.info(f"New estimate: {new_avg_ips}")
-                    # self.pending_iters[challenge_hash].remove(iterations_needed)
+                self.pending_iters[challenge_hash].remove(iterations_needed)
             else:
                 log.info(
                     f"Finished PoT chall:{challenge_hash[:10].hex()}.. {iterations_needed}"
@@ -180,6 +180,31 @@ async def _update_proofs_count(self, challenge_weight):
                         del self.active_discriminants_start_time[active_disc]
                         self.done_discriminants.append(active_disc)
 
+    async def _send_iterations(self, challenge_hash, writer):
+        alive_discriminant = True
+        while (alive_discriminant):
+            async with self.lock:
+                if (challenge_hash in self.active_discriminants) and (
+                    challenge_hash in self.pending_iters
+                ):
+                    if challenge_hash not in self.submitted_iters:
+                        self.submitted_iters[challenge_hash] = []
+                    for iter in sorted(self.pending_iters[challenge_hash]):
+                        if iter in self.submitted_iters[challenge_hash]:
+                            continue
+                        self.submitted_iters[challenge_hash].append(iter)
+                        if len(str(iter)) < 10:
+                            iter_size = "0" + str(len(str(iter)))
+                        else:
+                            iter_size = str(len(str(iter)))
+                        writer.write((iter_size + str(iter)).encode())
+                        await writer.drain()
+                        log.info(f"New iteration submitted: {iter}")
+            await asyncio.sleep(3)
+            async with self.lock:
+                if (challenge_hash in self.done_discriminants):
+                    alive_discriminant = False
+
     async def _do_process_communication(
         self, challenge_hash, challenge_weight, ip, port
     ):
@@ -219,29 +244,10 @@ async def _do_process_communication(
             self.active_discriminants[challenge_hash] = (writer, challenge_weight, ip)
             self.active_discriminants_start_time[challenge_hash] = time.time()
 
+        asyncio.create_task(self._send_iterations(challenge_hash, writer))
+
         # Listen to the server until "STOP" is received.
         while True:
-            async with self.lock:
-                if (challenge_hash in self.active_discriminants) and (
-                    challenge_hash in self.pending_iters
-                ):
-                    if challenge_hash not in self.submitted_iters:
-                        self.submitted_iters[challenge_hash] = []
-                    log.info(
-                        f"Pending: {self.pending_iters[challenge_hash]} "
-                        f"Submitted: {self.submitted_iters[challenge_hash]} Hash: {challenge_hash}"
-                    )
-                    for iter in sorted(self.pending_iters[challenge_hash]):
-                        if iter in self.submitted_iters[challenge_hash]:
-                            continue
-                        self.submitted_iters[challenge_hash].append(iter)
-                        if len(str(iter)) < 10:
-                            iter_size = "0" + str(len(str(iter)))
-                        else:
-                            iter_size = str(len(str(iter)))
-                        writer.write((iter_size + str(iter)).encode())
-                        await writer.drain()
-
             try:
                 data = await reader.readexactly(4)
             except (asyncio.IncompleteReadError, ConnectionResetError) as e:
@@ -259,21 +265,6 @@ async def _do_process_communication(
                     len_server = len(self.free_servers)
                     log.info(f"Process ended... Server length {len_server}")
                 break
-            elif data.decode() == "POLL":
-                async with self.lock:
-                    # If I have a newer discriminant... Free up the VDF server
-                    if (
-                        len(self.discriminant_queue) > 0
-                        and challenge_weight
-                        < max([h for _, h in self.discriminant_queue])
-                        and challenge_hash in self.active_discriminants
-                    ):
-                        log.info("Got poll, stopping the challenge!")
-                        writer.write(b"010")
-                        await writer.drain()
-                        del self.active_discriminants[challenge_hash]
-                        del self.active_discriminants_start_time[challenge_hash]
-                        self.done_discriminants.append(challenge_hash)
             else:
                 try:
                     # This must be a proof, read the continuation.
@@ -341,9 +332,25 @@ async def _manage_discriminant_queue(self):
                         )
                         self.discriminant_queue.clear()
                     else:
-                        disc = next(
+                        max_weight_disc = [
                             d for d, h in self.discriminant_queue if h == max_weight
-                        )
+                        ]
+                        with_iters = [
+                            d for d in max_weight_disc
+                            if d in self.pending_iters
+                            and len(self.pending_iters[d]) != 0
+                        ]
+                        if (len(with_iters) == 0):
+                            disc = max_weight_disc[0]
+                        else:
+                            min_iter = min([
+                                min(self.pending_iters[d])
+                                for d in with_iters
+                            ])
+                            disc = next(
+                                d for d in with_iters
+                                if min(self.pending_iters[d]) == min_iter
+                            )
                         if len(self.free_servers) != 0:
                             ip, port = self.free_servers[0]
                             self.free_servers = self.free_servers[1:]
@@ -415,10 +422,14 @@ async def proof_of_space_info(
 
             if proof_of_space_info.challenge_hash not in self.pending_iters:
                 self.pending_iters[proof_of_space_info.challenge_hash] = []
+            if proof_of_space_info.challenge_hash not in self.submitted_iters:
+                self.submitted_iters[proof_of_space_info.challenge_hash] = []
 
             if (
                 proof_of_space_info.iterations_needed
                 not in self.pending_iters[proof_of_space_info.challenge_hash]
+                and proof_of_space_info.iterations_needed
+                not in self.submitted_iters[proof_of_space_info.challenge_hash]
             ):
                 log.info(
                     f"proof_of_space_info {proof_of_space_info.challenge_hash} adding "