body fetching flow control

Author: rairyx

pyethapp/synchronizer.py

@@ -146,7 +146,6 @@ def fetch_hashchain(self):
 
             if skeleton_fetch and not skeleton:
                remaining = self.fetch_headers(self.skeleton_peer,from0)
-                
                skeleton_fetch = False 
                if not remaining:
                   log_st.warn('no more skeleton received')
@@ -155,20 +154,18 @@ def fetch_hashchain(self):
                #should not continuew??
 
             if skeleton_fetch:    
-            
                self.fetch_headerbatch(from0,skeleton)
-               log_st.debug('header batch', headerbatch= self.batch_result) 
+            #   log_st.debug('header batch', headerbatch= self.batch_result) 
               # check result
                if self.header_processed>0 :
-         #          self.fetch_blocks(header_batch)
-                   #from0 = from0 + self.max_skeleton_size*self.max_blockheaders_per_request   
                    from0 += self.header_processed
+                   remaining =self.batch_result[self.header_processed:]    
                else:
                    return self.exit(success=False)
-          
+            #scheduling block download for unprocessed headers in the skeleton or last header batch 
             if remaining:
-                log_st.debug('fetching new skeletons')
-             #   self.fetch_blocks(remaining)
+                log_st.debug('scheduling new headers',count= len(remaining), start=from0)
+                self.synchronizer.blockheader_queue.put(remaining)
                 from0+=len(remaining)
 
 
@@ -233,7 +230,7 @@ def fetch_headerbatch(self,origin,skeleton):
                  self.pending_headerRequests[proto] = HeaderRequest(start) 
                  proto.idle = False 
                  fetching = True
-                 log_st.debug('sent header request',request= start , proto=proto)
+                 log_st.debug('sent header request',request=start , proto=proto)
              else:
                  task_empty= True
                  break
@@ -331,15 +328,19 @@ def verify_headers(self,proto,headers):
 
     def fetch_headers(self,proto, fromx):
         deferred = AsyncResult()
-        blockheaders_batch=[]
-        proto.send_getblockheaders(fromx,self.max_blockheaders_per_request)
+        blockheaders_batch=None
+        proto.send_getblockheaders(fromx,self.max_blockheaders_per_request,0,1)
         try:
+            self.requests[proto] = deferred 
             blockheaders_batch = deferred.get(block=True,timeout=self.blockheaders_request_timeout)
         except gevent.Timeout:
             log_st.warn('syncing batch hashchain timed out')
-            return []
+            proto.stop()
+            return self.exit(success=False)
         finally:
-            return blockheaders_batch
+            del self.requests[proto]
+        
+        return blockheaders_batch
 
 
     def receive_blockheaders(self, proto, blockheaders):
@@ -405,7 +406,7 @@ def run(self):
     def schedule_block_fetch(self):
         batch_header = []
         log_st.debug('start sheduleing blocks')
-        self.synchronizer.blockheader_queue = Queue() 
+        self.synchronizer.blockheader_queue = Queue(maxsize=8192) 
 
         while True:
           batch_header= self.synchronizer.blockheader_queue.get()
@@ -436,32 +437,18 @@ def fetch_blocks(self):
         num_blocks = 0
         num_fetched = 0
         retry = 0
-        last_block = None 
-#        batch_header = []
-        #while not self.blockheaders_queue.empty():
-#        self.synchronizer.blockheader_queue = Queue() 
-#        while True:
-#          batch_header= self.synchronizer.blockheader_queue.get()
-#          num_blocks = len(batch_header)
-#          log_body_st.debug('delivered headers', delivered_heades=batch_header)
-#          gevent.sleep(0)
-#          while batch_header: 
-#            limit = len(batch_header) if len(batch_header) < self.max_blocks_process else self.max_blocks_process
-#            blockbody_batch = batch_header[:limit]
-#            for header in blockbody_batch:
-                #check chain order 
-                #self.block_requests_pool[header.hash]= header
-#                self.block_requests_pool.append(header)
-#                self.bodytask_queue.put((header.number,header))
-            #check if length block_requests_pool is equal to blockhashes_batch
-#            batch_header = batch_header[limit:]     
-
+        last_block = None
+        throttled = False
         while True:
           try:
                 result = self.fetch_ready.get()
                 log_st.debug('start fetching blocks')
+                if throttled:
+                   gevent.sleep(0.1)
+                num_blocks = len(self.block_requests_pool)   
                 deferred = AsyncResult()
                 self.body_request=deferred 
+             #   throttled = False
 
               #check timed out pending requests
                 for proto in list(self.pending_bodyRequests):
@@ -477,8 +464,6 @@ def fetch_blocks(self):
                      #     proto.body_idle = True   
                           proto.stop()
 
-             #   log_st.debug('header task queue size, pending queue size, batch_requestsize',size=self.bodytask_queue.qsize(),pending=len(self.pending_blockRequests),batch_request=len(self.block_requests_pool)) 
-              #if self.headertask_queue.qsize == 0 and len(self.pending_headerRequests)==0 and len(self.batch_requests)==0 :
                 if len(self.block_requests_pool) == 0:
                    log_body_st.debug('block body fetching completed!')
                   # return True
@@ -491,6 +476,14 @@ def fetch_blocks(self):
                #    assert proto not in self.requests
                    if proto.is_stopped:
                         continue
+                   
+                   #if self.chainservice.block_queue.qsize()> 1010:
+                   if pending>8192-num_fetched:
+                       throttled = True  
+                       log_body_st.debug('throttled')
+                       break
+                   else:
+                       throttled = False
                    proto_deferred = AsyncResult()
                    # check if it's finished
                    block_batch=[]
@@ -515,10 +508,10 @@ def fetch_blocks(self):
                      break
               # check if there are protos available for header fetching 
               #if not fetching and not self.headertask_queue.empty() and pending == len(self.pending_headerRequests):
-                if not fetching and not task_empty:
+                if not fetching and not task_empty and not throttled:
                      log_body_st.warn('no protocols available')
                      return self.exit(success=False) 
-                elif task_empty and pending==len(self.pending_bodyRequests):
+                if task_empty or throttled and pending==len(self.pending_bodyRequests): 
                      continue
                 try:
                       proto_received = deferred.get(timeout=self.blocks_request_timeout)['proto']
@@ -558,7 +551,7 @@ def fetch_blocks(self):
                 # add received t_blocks
                 num_fetched += len(bodies)
                 log_body_st.debug('received block bodies',
-                        num=len(bodies),blockbody=bodies,num_fetched=num_fetched,
+                        num=len(bodies),num_fetched=num_fetched,
                              total=num_blocks, missing=num_blocks - num_fetched)
                 proto_received.body_idle=True                                                                      
                 del self.requests[proto_received]
@@ -581,18 +574,6 @@ def fetch_blocks(self):
           except Exception as ex:
               log_body_st.error(error = ex)
 
-                
-
-        #        for body in bodies:
-        #              try:
-        #                h = headers.pop(0)
-        #                t_block = TransientBlock(h, body.transactions, body.uncles)
-        #                self.chainservice.add_block(t_block, proto)  # this blocks if the queue is full
-        #                self.block_requests_pool.remove(h) 
-        #              except IndexError as e:
-        #                log_st.error('headers and bodies mismatch', error=e)
-        #                self.exit(success=False)
-        #        log_st.debug('adding blocks done', took=time.time() - ts)
 
                  # done
         last_block = t_block
@@ -665,7 +646,7 @@ def __init__(self, chainservice, force_sync=None):
         self._protocols = dict()  # proto: chain_difficulty
         self.synctask = None
         self.syncbody = None
-        self.blockheader_queue = Queue() 
+        self.blockheader_queue = Queue(maxsize=8192) 
 
     def synctask_exited(self, success=False):
         # note: synctask broadcasts best block