Merge pull request #863 from karalabe/ordered-block-download

eth/downloader: prioritize block fetch based on chain position, cap memo...

Godeps/Godeps.json

@@ -98,6 +98,10 @@
 			"Comment": "v0.1.0-3-g27c4092",
 			"Rev": "27c40922c40b43fe04554d8223a402af3ea333f3"
 		},
+		{
+			"ImportPath": "gopkg.in/karalabe/cookiejar.v2/collections/prque",
+			"Rev": "0b2e270613f5d7ba262a5749b9e32270131497a2"
+		},
 		{
 			"ImportPath": "gopkg.in/qml.v1/cdata",
 			"Rev": "1116cb9cd8dee23f8d444ded354eb53122739f99"

Godeps/_workspace/src/gopkg.in/karalabe/cookiejar.v2/collections/prque/example_test.go

+// CookieJar - A contestant's algorithm toolbox
+// Copyright (c) 2013 Peter Szilagyi. All rights reserved.
+//
+// CookieJar is dual licensed: you can redistribute it and/or modify it under
+// the terms of the GNU General Public License as published by the Free Software
+// Foundation, either version 3 of the License, or (at your option) any later
+// version.
+//
+// The toolbox is distributed in the hope that it will be useful, but WITHOUT
+// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+// FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
+// more details.
+//
+// Alternatively, the CookieJar toolbox may be used in accordance with the terms
+// and conditions contained in a signed written agreement between you and the
+// author(s).
+
+package prque_test
+
+import (
+	"fmt"
+
+	"gopkg.in/karalabe/cookiejar.v2/collections/prque"
+)
+
+// Insert some data into a priority queue and pop them out in prioritized order.
+func Example_usage() {
+	// Define some data to push into the priority queue
+	prio := []float32{77.7, 22.2, 44.4, 55.5, 11.1, 88.8, 33.3, 99.9, 0.0, 66.6}
+	data := []string{"zero", "one", "two", "three", "four", "five", "six", "seven", "eight", "nine"}
+
+	// Create the priority queue and insert the prioritized data
+	pq := prque.New()
+	for i := 0; i < len(data); i++ {
+		pq.Push(data[i], prio[i])
+	}
+	// Pop out the data and print them
+	for !pq.Empty() {
+		val, prio := pq.Pop()
+		fmt.Printf("%.1f:%s ", prio, val)
+	}
+	// Output:
+	// 99.9:seven 88.8:five 77.7:zero 66.6:nine 55.5:three 44.4:two 33.3:six 22.2:one 11.1:four 0.0:eight
+}

Godeps/_workspace/src/gopkg.in/karalabe/cookiejar.v2/collections/prque/prque.go

+// CookieJar - A contestant's algorithm toolbox
+// Copyright (c) 2013 Peter Szilagyi. All rights reserved.
+//
+// CookieJar is dual licensed: you can redistribute it and/or modify it under
+// the terms of the GNU General Public License as published by the Free Software
+// Foundation, either version 3 of the License, or (at your option) any later
+// version.
+//
+// The toolbox is distributed in the hope that it will be useful, but WITHOUT
+// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+// FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
+// more details.
+//
+// Alternatively, the CookieJar toolbox may be used in accordance with the terms
+// and conditions contained in a signed written agreement between you and the
+// author(s).
+
+// Package prque implements a priority queue data structure supporting arbitrary
+// value types and float priorities.
+//
+// The reasoning behind using floats for the priorities vs. ints or interfaces
+// was larger flexibility without sacrificing too much performance or code
+// complexity.
+//
+// If you would like to use a min-priority queue, simply negate the priorities.
+//
+// Internally the queue is based on the standard heap package working on a
+// sortable version of the block based stack.
+package prque
+
+import (
+	"container/heap"
+)
+
+// Priority queue data structure.
+type Prque struct {
+	cont *sstack
+}
+
+// Creates a new priority queue.
+func New() *Prque {
+	return &Prque{newSstack()}
+}
+
+// Pushes a value with a given priority into the queue, expanding if necessary.
+func (p *Prque) Push(data interface{}, priority float32) {
+	heap.Push(p.cont, &item{data, priority})
+}
+
+// Pops the value with the greates priority off the stack and returns it.
+// Currently no shrinking is done.
+func (p *Prque) Pop() (interface{}, float32) {
+	item := heap.Pop(p.cont).(*item)
+	return item.value, item.priority
+}
+
+// Pops only the item from the queue, dropping the associated priority value.
+func (p *Prque) PopItem() interface{} {
+	return heap.Pop(p.cont).(*item).value
+}
+
+// Checks whether the priority queue is empty.
+func (p *Prque) Empty() bool {
+	return p.cont.Len() == 0
+}
+
+// Returns the number of element in the priority queue.
+func (p *Prque) Size() int {
+	return p.cont.Len()
+}
+
+// Clears the contents of the priority queue.
+func (p *Prque) Reset() {
+	*p = *New()
+}

Godeps/_workspace/src/gopkg.in/karalabe/cookiejar.v2/collections/prque/prque_test.go

+// CookieJar - A contestant's algorithm toolbox
+// Copyright (c) 2013 Peter Szilagyi. All rights reserved.
+//
+// CookieJar is dual licensed: you can redistribute it and/or modify it under
+// the terms of the GNU General Public License as published by the Free Software
+// Foundation, either version 3 of the License, or (at your option) any later
+// version.
+//
+// The toolbox is distributed in the hope that it will be useful, but WITHOUT
+// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+// FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
+// more details.
+//
+// Alternatively, the CookieJar toolbox may be used in accordance with the terms
+// and conditions contained in a signed written agreement between you and the
+// author(s).
+
+package prque
+
+import (
+	"math/rand"
+	"testing"
+)
+
+func TestPrque(t *testing.T) {
+	// Generate a batch of random data and a specific priority order
+	size := 16 * blockSize
+	prio := rand.Perm(size)
+	data := make([]int, size)
+	for i := 0; i < size; i++ {
+		data[i] = rand.Int()
+	}
+	queue := New()
+	for rep := 0; rep < 2; rep++ {
+		// Fill a priority queue with the above data
+		for i := 0; i < size; i++ {
+			queue.Push(data[i], float32(prio[i]))
+			if queue.Size() != i+1 {
+				t.Errorf("queue size mismatch: have %v, want %v.", queue.Size(), i+1)
+			}
+		}
+		// Create a map the values to the priorities for easier verification
+		dict := make(map[float32]int)
+		for i := 0; i < size; i++ {
+			dict[float32(prio[i])] = data[i]
+		}
+		// Pop out the elements in priority order and verify them
+		prevPrio := float32(size + 1)
+		for !queue.Empty() {
+			val, prio := queue.Pop()
+			if prio > prevPrio {
+				t.Errorf("invalid priority order: %v after %v.", prio, prevPrio)
+			}
+			prevPrio = prio
+			if val != dict[prio] {
+				t.Errorf("push/pop mismatch: have %v, want %v.", val, dict[prio])
+			}
+			delete(dict, prio)
+		}
+	}
+}
+
+func TestReset(t *testing.T) {
+	// Generate a batch of random data and a specific priority order
+	size := 16 * blockSize
+	prio := rand.Perm(size)
+	data := make([]int, size)
+	for i := 0; i < size; i++ {
+		data[i] = rand.Int()
+	}
+	queue := New()
+	for rep := 0; rep < 2; rep++ {
+		// Fill a priority queue with the above data
+		for i := 0; i < size; i++ {
+			queue.Push(data[i], float32(prio[i]))
+			if queue.Size() != i+1 {
+				t.Errorf("queue size mismatch: have %v, want %v.", queue.Size(), i+1)
+			}
+		}
+		// Create a map the values to the priorities for easier verification
+		dict := make(map[float32]int)
+		for i := 0; i < size; i++ {
+			dict[float32(prio[i])] = data[i]
+		}
+		// Pop out half the elements in priority order and verify them
+		prevPrio := float32(size + 1)
+		for i := 0; i < size/2; i++ {
+			val, prio := queue.Pop()
+			if prio > prevPrio {
+				t.Errorf("invalid priority order: %v after %v.", prio, prevPrio)
+			}
+			prevPrio = prio
+			if val != dict[prio] {
+				t.Errorf("push/pop mismatch: have %v, want %v.", val, dict[prio])
+			}
+			delete(dict, prio)
+		}
+		// Reset and ensure it's empty
+		queue.Reset()
+		if !queue.Empty() {
+			t.Errorf("priority queue not empty after reset: %v", queue)
+		}
+	}
+}
+
+func BenchmarkPush(b *testing.B) {
+	// Create some initial data
+	data := make([]int, b.N)
+	prio := make([]float32, b.N)
+	for i := 0; i < len(data); i++ {
+		data[i] = rand.Int()
+		prio[i] = rand.Float32()
+	}
+	// Execute the benchmark
+	b.ResetTimer()
+	queue := New()
+	for i := 0; i < len(data); i++ {
+		queue.Push(data[i], prio[i])
+	}
+}
+
+func BenchmarkPop(b *testing.B) {
+	// Create some initial data
+	data := make([]int, b.N)
+	prio := make([]float32, b.N)
+	for i := 0; i < len(data); i++ {
+		data[i] = rand.Int()
+		prio[i] = rand.Float32()
+	}
+	queue := New()
+	for i := 0; i < len(data); i++ {
+		queue.Push(data[i], prio[i])
+	}
+	// Execute the benchmark
+	b.ResetTimer()
+	for !queue.Empty() {
+		queue.Pop()
+	}
+}

Godeps/_workspace/src/gopkg.in/karalabe/cookiejar.v2/collections/prque/sstack.go

+// CookieJar - A contestant's algorithm toolbox
+// Copyright (c) 2013 Peter Szilagyi. All rights reserved.
+//
+// CookieJar is dual licensed: you can redistribute it and/or modify it under
+// the terms of the GNU General Public License as published by the Free Software
+// Foundation, either version 3 of the License, or (at your option) any later
+// version.
+//
+// The toolbox is distributed in the hope that it will be useful, but WITHOUT
+// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+// FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
+// more details.
+//
+// Alternatively, the CookieJar toolbox may be used in accordance with the terms
+// and conditions contained in a signed written agreement between you and the
+// author(s).
+
+package prque
+
+// The size of a block of data
+const blockSize = 4096
+
+// A prioritized item in the sorted stack.
+type item struct {
+	value    interface{}
+	priority float32
+}
+
+// Internal sortable stack data structure. Implements the Push and Pop ops for
+// the stack (heap) functionality and the Len, Less and Swap methods for the
+// sortability requirements of the heaps.
+type sstack struct {
+	size     int
+	capacity int
+	offset   int
+
+	blocks [][]*item
+	active []*item
+}
+
+// Creates a new, empty stack.
+func newSstack() *sstack {
+	result := new(sstack)
+	result.active = make([]*item, blockSize)
+	result.blocks = [][]*item{result.active}
+	result.capacity = blockSize
+	return result
+}
+
+// Pushes a value onto the stack, expanding it if necessary. Required by
+// heap.Interface.
+func (s *sstack) Push(data interface{}) {
+	if s.size == s.capacity {
+		s.active = make([]*item, blockSize)
+		s.blocks = append(s.blocks, s.active)
+		s.capacity += blockSize
+		s.offset = 0
+	} else if s.offset == blockSize {
+		s.active = s.blocks[s.size/blockSize]
+		s.offset = 0
+	}
+	s.active[s.offset] = data.(*item)
+	s.offset++
+	s.size++
+}
+
+// Pops a value off the stack and returns it. Currently no shrinking is done.
+// Required by heap.Interface.
+func (s *sstack) Pop() (res interface{}) {
+	s.size--
+	s.offset--
+	if s.offset < 0 {
+		s.offset = blockSize - 1
+		s.active = s.blocks[s.size/blockSize]
+	}
+	res, s.active[s.offset] = s.active[s.offset], nil
+	return
+}
+
+// Returns the length of the stack. Required by sort.Interface.
+func (s *sstack) Len() int {
+	return s.size
+}
+
+// Compares the priority of two elements of the stack (higher is first).
+// Required by sort.Interface.
+func (s *sstack) Less(i, j int) bool {
+	return s.blocks[i/blockSize][i%blockSize].priority > s.blocks[j/blockSize][j%blockSize].priority
+}
+
+// Swaps two elements in the stack. Required by sort.Interface.
+func (s *sstack) Swap(i, j int) {
+	ib, io, jb, jo := i/blockSize, i%blockSize, j/blockSize, j%blockSize
+	s.blocks[ib][io], s.blocks[jb][jo] = s.blocks[jb][jo], s.blocks[ib][io]
+}
+
+// Resets the stack, effectively clearing its contents.
+func (s *sstack) Reset() {
+	*s = *newSstack()
+}

Godeps/_workspace/src/gopkg.in/karalabe/cookiejar.v2/collections/prque/sstack_test.go

+// CookieJar - A contestant's algorithm toolbox
+// Copyright (c) 2013 Peter Szilagyi. All rights reserved.
+//
+// CookieJar is dual licensed: you can redistribute it and/or modify it under
+// the terms of the GNU General Public License as published by the Free Software
+// Foundation, either version 3 of the License, or (at your option) any later
+// version.
+//
+// The toolbox is distributed in the hope that it will be useful, but WITHOUT
+// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+// FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
+// more details.
+//
+// Alternatively, the CookieJar toolbox may be used in accordance with the terms
+// and conditions contained in a signed written agreement between you and the
+// author(s).
+
+package prque
+
+import (
+	"math/rand"
+	"sort"
+	"testing"
+)
+
+func TestSstack(t *testing.T) {
+	// Create some initial data
+	size := 16 * blockSize
+	data := make([]*item, size)
+	for i := 0; i < size; i++ {
+		data[i] = &item{rand.Int(), rand.Float32()}
+	}
+	stack := newSstack()
+	for rep := 0; rep < 2; rep++ {
+		// Push all the data into the stack, pop out every second
+		secs := []*item{}
+		for i := 0; i < size; i++ {
+			stack.Push(data[i])
+			if i%2 == 0 {
+				secs = append(secs, stack.Pop().(*item))
+			}
+		}
+		rest := []*item{}
+		for stack.Len() > 0 {
+			rest = append(rest, stack.Pop().(*item))
+		}
+		// Make sure the contents of the resulting slices are ok
+		for i := 0; i < size; i++ {
+			if i%2 == 0 && data[i] != secs[i/2] {
+				t.Errorf("push/pop mismatch: have %v, want %v.", secs[i/2], data[i])
+			}
+			if i%2 == 1 && data[i] != rest[len(rest)-i/2-1] {
+				t.Errorf("push/pop mismatch: have %v, want %v.", rest[len(rest)-i/2-1], data[i])
+			}
+		}
+	}
+}
+
+func TestSstackSort(t *testing.T) {
+	// Create some initial data
+	size := 16 * blockSize
+	data := make([]*item, size)
+	for i := 0; i < size; i++ {
+		data[i] = &item{rand.Int(), float32(i)}
+	}
+	// Push all the data into the stack
+	stack := newSstack()
+	for _, val := range data {
+		stack.Push(val)
+	}
+	// Sort and pop the stack contents (should reverse the order)
+	sort.Sort(stack)
+	for _, val := range data {
+		out := stack.Pop()
+		if out != val {
+			t.Errorf("push/pop mismatch after sort: have %v, want %v.", out, val)
+		}
+	}
+}
+
+func TestSstackReset(t *testing.T) {
+	// Create some initial data
+	size := 16 * blockSize
+	data := make([]*item, size)
+	for i := 0; i < size; i++ {
+		data[i] = &item{rand.Int(), rand.Float32()}
+	}
+	stack := newSstack()
+	for rep := 0; rep < 2; rep++ {
+		// Push all the data into the stack, pop out every second
+		secs := []*item{}
+		for i := 0; i < size; i++ {
+			stack.Push(data[i])
+			if i%2 == 0 {
+				secs = append(secs, stack.Pop().(*item))
+			}
+		}
+		// Reset and verify both pulled and stack contents
+		stack.Reset()
+		if stack.Len() != 0 {
+			t.Errorf("stack not empty after reset: %v", stack)
+		}
+		for i := 0; i < size; i++ {
+			if i%2 == 0 && data[i] != secs[i/2] {
+				t.Errorf("push/pop mismatch: have %v, want %v.", secs[i/2], data[i])
+			}
+		}
+	}
+}

eth/downloader/downloader_test.go

@@ -128,7 +128,7 @@ func TestDownload(t *testing.T) {
 		t.Error("download error", err)
 	}
 
-	inqueue := len(tester.downloader.queue.blocks)
+	inqueue := len(tester.downloader.queue.blockCache)
 	if inqueue != targetBlocks {
 		t.Error("expected", targetBlocks, "have", inqueue)
 	}
@@ -151,7 +151,7 @@ func TestMissing(t *testing.T) {
 		t.Error("download error", err)
 	}
 
-	inqueue := len(tester.downloader.queue.blocks)
+	inqueue := len(tester.downloader.queue.blockCache)
 	if inqueue != targetBlocks {
 		t.Error("expected", targetBlocks, "have", inqueue)
 	}
@@ -181,3 +181,51 @@ func TestTaking(t *testing.T) {
 		t.Error("expected to take 1000, got", len(bs1))
 	}
 }
+
+func TestThrottling(t *testing.T) {
+	minDesiredPeerCount = 4
+	blockTtl = 1 * time.Second
+
+	targetBlocks := 4 * blockCacheLimit
+	hashes := createHashes(0, targetBlocks)
+	blocks := createBlocksFromHashes(hashes)
+	tester := newTester(t, hashes, blocks)
+
+	tester.newPeer("peer1", big.NewInt(10000), hashes[0])
+	tester.newPeer("peer2", big.NewInt(0), common.Hash{})
+	tester.badBlocksPeer("peer3", big.NewInt(0), common.Hash{})
+	tester.badBlocksPeer("peer4", big.NewInt(0), common.Hash{})
+
+	// Concurrently download and take the blocks
+	errc := make(chan error, 1)
+	go func() {
+		errc <- tester.sync("peer1", hashes[0])
+	}()
+
+	done := make(chan struct{})
+	took := []*types.Block{}
+	go func() {
+		for {
+			select {
+			case <-done:
+				took = append(took, tester.downloader.TakeBlocks()...)
+				done <- struct{}{}
+				return
+			default:
+				took = append(took, tester.downloader.TakeBlocks()...)
+			}
+		}
+	}()
+
+	// Synchronise the two threads and verify
+	err := <-errc
+	done <- struct{}{}
+	<-done
+
+	if err != nil {
+		t.Fatalf("failed to synchronise blocks: %v", err)
+	}
+	if len(took) != targetBlocks {
+		t.Fatalf("downloaded block mismatch: have %v, want %v", len(took), targetBlocks)
+	}
+}

eth/downloader/peer.go

@@ -78,7 +78,7 @@ func newPeer(id string, hash common.Hash, getHashes hashFetcherFn, getBlocks blo
 }
 
 // fetch a chunk using the peer
-func (p *peer) fetch(chunk *chunk) error {
+func (p *peer) fetch(request *fetchRequest) error {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
@@ -88,13 +88,12 @@ func (p *peer) fetch(chunk *chunk) error {
 
 	// set working state
 	p.state = workingState
-	// convert the set to a fetchable slice
-	hashes, i := make([]common.Hash, chunk.hashes.Size()), 0
-	chunk.hashes.Each(func(v interface{}) bool {
-		hashes[i] = v.(common.Hash)
-		i++
-		return true
-	})
+
+	// Convert the hash set to a fetchable slice
+	hashes := make([]common.Hash, 0, len(request.Hashes))
+	for hash, _ := range request.Hashes {
+		hashes = append(hashes, hash)
+	}
 	p.getBlocks(hashes)
 
 	return nil

eth/downloader/queue_test.go

@@ -32,31 +32,30 @@ func createBlocksFromHashSet(hashes *set.Set) []*types.Block {
 }
 
 func TestChunking(t *testing.T) {
-	queue := newqueue()
+	queue := newQueue()
 	peer1 := newPeer("peer1", common.Hash{}, nil, nil)
 	peer2 := newPeer("peer2", common.Hash{}, nil, nil)
 
 	// 99 + 1 (1 == known genesis hash)
 	hashes := createHashes(0, 99)
-	hashSet := createHashSet(hashes)
-	queue.put(hashSet)
+	queue.Insert(hashes)
 
-	chunk1 := queue.get(peer1, 99)
+	chunk1 := queue.Reserve(peer1, 99)
 	if chunk1 == nil {
 		t.Errorf("chunk1 is nil")
 		t.FailNow()
 	}
-	chunk2 := queue.get(peer2, 99)
+	chunk2 := queue.Reserve(peer2, 99)
 	if chunk2 == nil {
 		t.Errorf("chunk2 is nil")
 		t.FailNow()
 	}
 
-	if chunk1.hashes.Size() != 99 {
-		t.Error("expected chunk1 hashes to be 99, got", chunk1.hashes.Size())
+	if len(chunk1.Hashes) != 99 {
+		t.Error("expected chunk1 hashes to be 99, got", len(chunk1.Hashes))
 	}
 
-	if chunk2.hashes.Size() != 1 {
-		t.Error("expected chunk1 hashes to be 1, got", chunk2.hashes.Size())
+	if len(chunk2.Hashes) != 1 {
+		t.Error("expected chunk1 hashes to be 1, got", len(chunk2.Hashes))
 	}
 }

eth/handler.go

@@ -19,9 +19,9 @@ import (
 )
 
 const (
-	peerCountTimeout    = 12 * time.Second // Amount of time it takes for the peer handler to ignore minDesiredPeerCount
-	blockProcTimer      = 500 * time.Millisecond
-	minDesiredPeerCount = 5 // Amount of peers desired to start syncing
+	forceSyncCycle      = 10 * time.Second       // Time interval to force syncs, even if few peers are available
+	blockProcCycle      = 500 * time.Millisecond // Time interval to check for new blocks to process
+	minDesiredPeerCount = 5                      // Amount of peers desired to start syncing
 	blockProcAmount     = 256
 )
 
@@ -307,7 +307,7 @@ func (self *ProtocolManager) handleMsg(p *peer) error {
 
 		// Attempt to insert the newly received by checking if the parent exists.
 		// if the parent exists we process the block and propagate to our peers
-		// otherwise synchronise with the peer
+		// otherwise synchronize with the peer
 		if self.chainman.HasBlock(request.Block.ParentHash()) {
 			if _, err := self.chainman.InsertChain(types.Blocks{request.Block}); err != nil {
 				glog.V(logger.Error).Infoln("removed peer (", p.id, ") due to block error")

eth/sync.go

@@ -12,10 +12,8 @@ import (
 // Sync contains all synchronisation code for the eth protocol
 
 func (pm *ProtocolManager) update() {
-	// itimer is used to determine when to start ignoring `minDesiredPeerCount`
-	itimer := time.NewTimer(peerCountTimeout)
-	// btimer is used for picking of blocks from the downloader
-	btimer := time.Tick(blockProcTimer)
+	forceSync := time.Tick(forceSyncCycle)
+	blockProc := time.Tick(blockProcCycle)
 
 	for {
 		select {
@@ -24,27 +22,22 @@ func (pm *ProtocolManager) update() {
 			if len(pm.peers) < minDesiredPeerCount {
 				break
 			}
-
-			// Find the best peer
+			// Find the best peer and synchronise with it
 			peer := getBestPeer(pm.peers)
 			if peer == nil {
-				glog.V(logger.Debug).Infoln("Sync attempt cancelled. No peers available")
+				glog.V(logger.Debug).Infoln("Sync attempt canceled. No peers available")
 			}
-
-			itimer.Stop()
 			go pm.synchronise(peer)
-		case <-itimer.C:
-			// The timer will make sure that the downloader keeps an active state
-			// in which it attempts to always check the network for highest td peers
-			// Either select the peer or restart the timer if no peers could
-			// be selected.
+
+		case <-forceSync:
+			// Force a sync even if not enough peers are present
 			if peer := getBestPeer(pm.peers); peer != nil {
 				go pm.synchronise(peer)
-			} else {
-				itimer.Reset(5 * time.Second)
 			}
-		case <-btimer:
+		case <-blockProc:
+			// Try to pull some blocks from the downloaded
 			go pm.processBlocks()
+
 		case <-pm.quitSync:
 			return
 		}
@@ -59,12 +52,11 @@ func (pm *ProtocolManager) processBlocks() error {
 	pm.wg.Add(1)
 	defer pm.wg.Done()
 
+	// Take a batch of blocks (will return nil if a previous batch has not reached the chain yet)
 	blocks := pm.downloader.TakeBlocks()
 	if len(blocks) == 0 {
 		return nil
 	}
-	defer pm.downloader.Done()
-
 	glog.V(logger.Debug).Infof("Inserting chain with %d blocks (#%v - #%v)\n", len(blocks), blocks[0].Number(), blocks[len(blocks)-1].Number())
 
 	for len(blocks) != 0 && !pm.quit {
@@ -83,26 +75,28 @@ func (pm *ProtocolManager) synchronise(peer *peer) {
 	if peer.td.Cmp(pm.chainman.Td()) <= 0 {
 		return
 	}
-	// Check downloader if it's busy so it doesn't show the sync message
-	// for every attempty
-	if pm.downloader.IsBusy() {
-		return
-	}
-
 	// FIXME if we have the hash in our chain and the TD of the peer is
 	// much higher than ours, something is wrong with us or the peer.
 	// Check if the hash is on our own chain
 	if pm.chainman.HasBlock(peer.recentHash) {
 		return
 	}
-
 	// Get the hashes from the peer (synchronously)
+	glog.V(logger.Debug).Infof("Attempting synchronisation: %v, 0x%x", peer.id, peer.recentHash)
+
 	err := pm.downloader.Synchronise(peer.id, peer.recentHash)
-	if err != nil && err == downloader.ErrBadPeer {
-		glog.V(logger.Debug).Infoln("removed peer from peer set due to bad action")
+	switch err {
+	case nil:
+		glog.V(logger.Debug).Infof("Synchronisation completed")
+
+	case downloader.ErrBusy:
+		glog.V(logger.Debug).Infof("Synchronisation already in progress")
+
+	case downloader.ErrTimeout:
+		glog.V(logger.Debug).Infof("Removing peer %v due to sync timeout", peer.id)
 		pm.removePeer(peer)
-	} else if err != nil {
-		// handle error
-		glog.V(logger.Detail).Infoln("error downloading:", err)
+
+	default:
+		glog.V(logger.Warn).Infof("Synchronisation failed: %v", err)
 	}
 }