Merge pull request #2149

158c3ecf core: thread most of handle_incoming_tx (moneromooo-monero)
f57ee382 cryptonote_protocol: retry stale spans early (moneromooo-monero)
90df52e1 cryptonote_protocol: light cleanup (moneromooo-monero)
84e23156 cryptonote_protocol: avoid spurious SYNCHRONIZED OK messages (moneromooo-monero)
5be43fcd cryptonote_protocol_handler: sync speedup (moneromooo-monero)

6 files changed, 1012 insertions, 159 deletions

diff --git a/src/cryptonote_protocol/block_queue.cpp b/src/cryptonote_protocol/block_queue.cpp
new file mode 100644
index 000000000..583c3abf4
--- /dev/null
+++ b/src/cryptonote_protocol/block_queue.cpp
@@ -0,0 +1,434 @@
+// Copyright (c) 2017, The Monero Project
+// 
+// All rights reserved.
+// 
+// Redistribution and use in source and binary forms, with or without modification, are
+// permitted provided that the following conditions are met:
+// 
+// 1. Redistributions of source code must retain the above copyright notice, this list of
+//    conditions and the following disclaimer.
+// 
+// 2. Redistributions in binary form must reproduce the above copyright notice, this list
+//    of conditions and the following disclaimer in the documentation and/or other
+//    materials provided with the distribution.
+// 
+// 3. Neither the name of the copyright holder nor the names of its contributors may be
+//    used to endorse or promote products derived from this software without specific
+//    prior written permission.
+// 
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
+// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
+// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+// 
+// Parts of this file are originally copyright (c) 2012-2013 The Cryptonote developers
+
+#include <vector>
+#include <unordered_map>
+#include <boost/uuid/nil_generator.hpp>
+#include "cryptonote_protocol_defs.h"
+#include "block_queue.h"
+
+#undef MONERO_DEFAULT_LOG_CATEGORY
+#define MONERO_DEFAULT_LOG_CATEGORY "cn.block_queue"
+
+namespace std {
+  static_assert(sizeof(size_t) <= sizeof(boost::uuids::uuid), "boost::uuids::uuid too small");
+  template<> struct hash<boost::uuids::uuid> {
+    std::size_t operator()(const boost::uuids::uuid &_v) const {
+      return reinterpret_cast<const std::size_t &>(_v);
+    }
+  };
+}
+
+namespace cryptonote
+{
+
+void block_queue::add_blocks(uint64_t height, std::list<cryptonote::block_complete_entry> bcel, const boost::uuids::uuid &connection_id, float rate, size_t size)
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+  remove_span(height);
+  blocks.insert(span(height, std::move(bcel), connection_id, rate, size));
+}
+
+void block_queue::add_blocks(uint64_t height, uint64_t nblocks, const boost::uuids::uuid &connection_id, boost::posix_time::ptime time)
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+  blocks.insert(span(height, nblocks, connection_id, time));
+}
+
+void block_queue::flush_spans(const boost::uuids::uuid &connection_id, bool all)
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+  block_map::iterator i = blocks.begin();
+  while (i != blocks.end())
+  {
+    block_map::iterator j = i++;
+    if (j->connection_id == connection_id && (all || j->blocks.size() == 0))
+    {
+      blocks.erase(j);
+    }
+  }
+}
+
+void block_queue::flush_stale_spans(const std::set<boost::uuids::uuid> &live_connections)
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+  block_map::iterator i = blocks.begin();
+  if (i != blocks.end() && is_blockchain_placeholder(*i))
+    ++i;
+  while (i != blocks.end())
+  {
+    block_map::iterator j = i++;
+    if (live_connections.find(j->connection_id) == live_connections.end() && j->blocks.size() == 0)
+    {
+      blocks.erase(j);
+    }
+  }
+}
+
+void block_queue::remove_span(uint64_t start_block_height)
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+  for (block_map::iterator i = blocks.begin(); i != blocks.end(); ++i)
+  {
+    if (i->start_block_height == start_block_height)
+    {
+      blocks.erase(i);
+      return;
+    }
+  }
+}
+
+void block_queue::remove_spans(const boost::uuids::uuid &connection_id, uint64_t start_block_height)
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+  for (block_map::iterator i = blocks.begin(); i != blocks.end(); )
+  {
+    block_map::iterator j = i++;
+    if (j->connection_id == connection_id && j->start_block_height <= start_block_height)
+    {
+      blocks.erase(j);
+    }
+  }
+}
+
+void block_queue::mark_last_block(uint64_t last_block_height)
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+  if (!blocks.empty() && is_blockchain_placeholder(*blocks.begin()))
+    blocks.erase(*blocks.begin());
+  for (block_map::iterator i = blocks.begin(); i != blocks.end(); )
+  {
+    block_map::iterator j = i++;
+    if (j->start_block_height + j->nblocks - 1 <= last_block_height)
+    {
+      blocks.erase(j);
+    }
+  }
+
+  // mark the current state of the db (for a fresh db, it's just the genesis block)
+  add_blocks(0, last_block_height + 1, boost::uuids::nil_uuid());
+  MDEBUG("last blocked marked at " << last_block_height);
+}
+
+uint64_t block_queue::get_max_block_height() const
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+  uint64_t height = 0;
+  for (const auto &span: blocks)
+  {
+    const uint64_t h = span.start_block_height + span.nblocks - 1;
+    if (h > height)
+      height = h;
+  }
+  return height;
+}
+
+void block_queue::print() const
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+  MDEBUG("Block queue has " << blocks.size() << " spans");
+  for (const auto &span: blocks)
+    MDEBUG("  " << span.start_block_height << " - " << (span.start_block_height+span.nblocks-1) << " (" << span.nblocks << ") - " << (is_blockchain_placeholder(span) ? "blockchain" : span.blocks.empty() ? "scheduled" : "filled    ") << "  " << span.connection_id << " (" << ((unsigned)(span.rate*10/1024.f))/10.f << " kB/s)");
+}
+
+std::string block_queue::get_overview() const
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+  if (blocks.empty())
+    return "[]";
+  block_map::const_iterator i = blocks.begin();
+  std::string s = std::string("[") + std::to_string(i->start_block_height + i->nblocks - 1) + ":";
+  while (++i != blocks.end())
+    s += i->blocks.empty() ? "." : "o";
+  s += "]";
+  return s;
+}
+
+std::pair<uint64_t, uint64_t> block_queue::reserve_span(uint64_t first_block_height, uint64_t last_block_height, uint64_t max_blocks, const boost::uuids::uuid &connection_id, boost::posix_time::ptime time)
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+
+  if (last_block_height < first_block_height || max_blocks == 0)
+  {
+    MDEBUG("reserve_span: early out: first_block_height " << first_block_height << ", last_block_height " << last_block_height << ", max_blocks " << max_blocks);
+    return std::make_pair(0, 0);
+  }
+
+  uint64_t max_block_height = get_max_block_height();
+  if (last_block_height > max_block_height)
+    max_block_height = last_block_height;
+  if (max_block_height == 0)
+  {
+    MDEBUG("reserve_span: max_block_height is 0");
+    return std::make_pair(first_block_height, std::min(last_block_height - first_block_height + 1, max_blocks));
+  }
+
+  uint64_t base = 0, last_placeholder_block = 0;
+  bool has_placeholder = false;
+  block_map::const_iterator i = blocks.begin();
+  if (i != blocks.end() && is_blockchain_placeholder(*i))
+  {
+    base = i->start_block_height + i->nblocks;
+    last_placeholder_block = base - 1;
+    has_placeholder = true;
+    ++i;
+    for (block_map::const_iterator j = i; j != blocks.end(); ++j)
+    {
+      if (j->start_block_height < base)
+        base = j->start_block_height;
+    }
+  }
+  if (base > first_block_height)
+    base = first_block_height;
+
+  CHECK_AND_ASSERT_MES (base <= max_block_height + 1, std::make_pair(0, 0), "Blockchain placeholder larger than max block height");
+  std::vector<uint8_t> bitmap(max_block_height + 1 - base, 0);
+  MDEBUG("base " << base << ", last_placeholder_block " << (has_placeholder ? std::to_string(last_placeholder_block) : "none") << ", first_block_height " << first_block_height);
+  if (has_placeholder && last_placeholder_block >= base)
+    memset(bitmap.data(), 1, last_placeholder_block + 1 - base);
+  while (i != blocks.end())
+  {
+    CHECK_AND_ASSERT_MES (i->start_block_height >= base, std::make_pair(0, 0), "Span starts before blochckain placeholder");
+    memset(bitmap.data() + i->start_block_height - base, 1, i->nblocks);
+    ++i;
+  }
+
+  const uint8_t *ptr = (const uint8_t*)memchr(bitmap.data() + first_block_height - base, 0, bitmap.size() - (first_block_height - base));
+  if (!ptr)
+  {
+    MDEBUG("reserve_span: 0 not found in bitmap: " << first_block_height << " " << bitmap.size());
+    print();
+    return std::make_pair(0, 0);
+  }
+  uint64_t start_block_height = ptr - bitmap.data() + base;
+  if (start_block_height > last_block_height)
+  {
+    MDEBUG("reserve_span: start_block_height > last_block_height: " << start_block_height << " < " << last_block_height);
+    return std::make_pair(0, 0);
+  }
+  if (start_block_height + max_blocks - 1 < first_block_height)
+  {
+    MDEBUG("reserve_span: start_block_height + max_blocks - 1 < first_block_height: " << start_block_height << " + " << max_blocks << " - 1 < " << first_block_height);
+    return std::make_pair(0, 0);
+  }
+
+  uint64_t nblocks = 1;
+  while (start_block_height + nblocks <= last_block_height && nblocks < max_blocks && bitmap[start_block_height + nblocks - base] == 0)
+    ++nblocks;
+
+  MDEBUG("Reserving span " << start_block_height << " - " << (start_block_height + nblocks - 1) << " for " << connection_id);
+  add_blocks(start_block_height, nblocks, connection_id, time);
+  return std::make_pair(start_block_height, nblocks);
+}
+
+bool block_queue::is_blockchain_placeholder(const span &span) const
+{
+  static const boost::uuids::uuid uuid0 = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0};
+  return span.connection_id == uuid0;
+}
+
+std::pair<uint64_t, uint64_t> block_queue::get_start_gap_span() const
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+  if (blocks.empty())
+    return std::make_pair(0, 0);
+  block_map::const_iterator i = blocks.begin();
+  if (!is_blockchain_placeholder(*i))
+    return std::make_pair(0, 0);
+  uint64_t current_height = i->start_block_height + i->nblocks - 1;
+  ++i;
+  if (i == blocks.end())
+    return std::make_pair(0, 0);
+  uint64_t first_span_height = i->start_block_height;
+  if (first_span_height <= current_height + 1)
+    return std::make_pair(0, 0);
+  MDEBUG("Found gap at start of spans: last blockchain block height " << current_height << ", first span's block height " << first_span_height);
+  print();
+  return std::make_pair(current_height + 1, first_span_height - current_height - 1);
+}
+
+std::pair<uint64_t, uint64_t> block_queue::get_next_span_if_scheduled(std::list<crypto::hash> &hashes, boost::uuids::uuid &connection_id, boost::posix_time::ptime &time) const
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+  if (blocks.empty())
+    return std::make_pair(0, 0);
+  block_map::const_iterator i = blocks.begin();
+  if (is_blockchain_placeholder(*i))
+    ++i;
+  if (i == blocks.end())
+    return std::make_pair(0, 0);
+  if (!i->blocks.empty())
+    return std::make_pair(0, 0);
+  hashes = i->hashes;
+  connection_id = i->connection_id;
+  time = i->time;
+  return std::make_pair(i->start_block_height, i->nblocks);
+}
+
+void block_queue::set_span_hashes(uint64_t start_height, const boost::uuids::uuid &connection_id, std::list<crypto::hash> hashes)
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+  for (block_map::iterator i = blocks.begin(); i != blocks.end(); ++i)
+  {
+    if (i->start_block_height == start_height && i->connection_id == connection_id)
+    {
+      span s = *i;
+      blocks.erase(i);
+      s.hashes = std::move(hashes);
+      blocks.insert(s);
+      return;
+    }
+  }
+}
+
+bool block_queue::get_next_span(uint64_t &height, std::list<cryptonote::block_complete_entry> &bcel, boost::uuids::uuid &connection_id, bool filled) const
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+  if (blocks.empty())
+    return false;
+  block_map::const_iterator i = blocks.begin();
+  if (is_blockchain_placeholder(*i))
+    ++i;
+  for (; i != blocks.end(); ++i)
+  {
+    if (!filled || !i->blocks.empty())
+    {
+      height = i->start_block_height;
+      bcel = i->blocks;
+      connection_id = i->connection_id;
+      return true;
+    }
+  }
+  return false;
+}
+
+size_t block_queue::get_data_size() const
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+  size_t size = 0;
+  for (const auto &span: blocks)
+    size += span.size;
+  return size;
+}
+
+size_t block_queue::get_num_filled_spans_prefix() const
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+
+  if (blocks.empty())
+    return 0;
+  block_map::const_iterator i = blocks.begin();
+  if (is_blockchain_placeholder(*i))
+    ++i;
+  size_t size = 0;
+  while (i != blocks.end() && !i->blocks.empty())
+  {
+    ++i;
+    ++size;
+  }
+  return size;
+}
+
+size_t block_queue::get_num_filled_spans() const
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+  size_t size = 0;
+  for (const auto &span: blocks)
+  if (!span.blocks.empty())
+    ++size;
+  return size;
+}
+
+crypto::hash block_queue::get_last_known_hash() const
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+  crypto::hash hash = cryptonote::null_hash;
+  uint64_t highest_height = 0;
+  for (const auto &span: blocks)
+  {
+    uint64_t h = span.start_block_height + span.nblocks - 1;
+    if (h > highest_height && span.hashes.size() == span.nblocks)
+    {
+      hash = span.hashes.back();
+      highest_height = h;
+    }
+  }
+  return hash;
+}
+
+float block_queue::get_speed(const boost::uuids::uuid &connection_id) const
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+  std::unordered_map<boost::uuids::uuid, float> speeds;
+  for (const auto &span: blocks)
+  {
+    if (span.blocks.empty())
+      continue;
+    // note that the average below does not average over the whole set, but over the
+    // previous pseudo average and the latest rate: this gives much more importance
+    // to the latest measurements, which is fine here
+    std::unordered_map<boost::uuids::uuid, float>::iterator i = speeds.find(span.connection_id);
+    if (i == speeds.end())
+      speeds.insert(std::make_pair(span.connection_id, span.rate));
+    else
+      i->second = (i->second + span.rate) / 2;
+  }
+  float conn_rate = -1, best_rate = 0;
+  for (auto i: speeds)
+  {
+    if (i.first == connection_id)
+      conn_rate = i.second;
+    if (i.second > best_rate)
+      best_rate = i.second;
+  }
+
+  if (conn_rate <= 0)
+    return 1.0f; // not found, assume good speed
+  if (best_rate == 0)
+    return 1.0f; // everything dead ? Can't happen, but let's trap anyway
+
+  const float speed = conn_rate / best_rate;
+  MTRACE(" Relative speed for " << connection_id << ": " << speed << " (" << conn_rate << "/" << best_rate);
+  return speed;
+}
+
+bool block_queue::foreach(std::function<bool(const span&)> f, bool include_blockchain_placeholder) const
+{
+  boost::unique_lock<boost::recursive_mutex> lock(mutex);
+  block_map::const_iterator i = blocks.begin();
+  if (!include_blockchain_placeholder && i != blocks.end() && is_blockchain_placeholder(*i))
+    ++i;
+  while (i != blocks.end())
+    if (!f(*i++))
+      return false;
+  return true;
+}
+
+}
diff --git a/src/cryptonote_protocol/block_queue.h b/src/cryptonote_protocol/block_queue.h
new file mode 100644
index 000000000..c75ebc0b9
--- /dev/null
+++ b/src/cryptonote_protocol/block_queue.h
@@ -0,0 +1,97 @@
+// Copyright (c) 2017, The Monero Project
+// 
+// All rights reserved.
+// 
+// Redistribution and use in source and binary forms, with or without modification, are
+// permitted provided that the following conditions are met:
+// 
+// 1. Redistributions of source code must retain the above copyright notice, this list of
+//    conditions and the following disclaimer.
+// 
+// 2. Redistributions in binary form must reproduce the above copyright notice, this list
+//    of conditions and the following disclaimer in the documentation and/or other
+//    materials provided with the distribution.
+// 
+// 3. Neither the name of the copyright holder nor the names of its contributors may be
+//    used to endorse or promote products derived from this software without specific
+//    prior written permission.
+// 
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
+// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
+// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+// 
+// Parts of this file are originally copyright (c) 2012-2013 The Cryptonote developers
+
+#pragma once
+
+#include <string>
+#include <list>
+#include <set>
+#include <boost/thread/recursive_mutex.hpp>
+#include <boost/uuid/uuid.hpp>
+
+#undef MONERO_DEFAULT_LOG_CATEGORY
+#define MONERO_DEFAULT_LOG_CATEGORY "cn.block_queue"
+
+namespace cryptonote
+{
+  struct block_complete_entry;
+
+  class block_queue
+  {
+  public:
+    struct span
+    {
+      uint64_t start_block_height;
+      std::list<crypto::hash> hashes;
+      std::list<cryptonote::block_complete_entry> blocks;
+      boost::uuids::uuid connection_id;
+      uint64_t nblocks;
+      float rate;
+      size_t size;
+      boost::posix_time::ptime time;
+
+      span(uint64_t start_block_height, std::list<cryptonote::block_complete_entry> blocks, const boost::uuids::uuid &connection_id, float rate, size_t size):
+        start_block_height(start_block_height), blocks(std::move(blocks)), connection_id(connection_id), nblocks(this->blocks.size()), rate(rate), size(size), time() {}
+      span(uint64_t start_block_height, uint64_t nblocks, const boost::uuids::uuid &connection_id, boost::posix_time::ptime time):
+        start_block_height(start_block_height), connection_id(connection_id), nblocks(nblocks), rate(0.0f), size(0), time(time) {}
+
+      bool operator<(const span &s) const { return start_block_height < s.start_block_height; }
+    };
+    typedef std::set<span> block_map;
+
+  public:
+    void add_blocks(uint64_t height, std::list<cryptonote::block_complete_entry> bcel, const boost::uuids::uuid &connection_id, float rate, size_t size);
+    void add_blocks(uint64_t height, uint64_t nblocks, const boost::uuids::uuid &connection_id, boost::posix_time::ptime time = boost::date_time::min_date_time);
+    void flush_spans(const boost::uuids::uuid &connection_id, bool all = false);
+    void flush_stale_spans(const std::set<boost::uuids::uuid> &live_connections);
+    void remove_span(uint64_t start_block_height);
+    void remove_spans(const boost::uuids::uuid &connection_id, uint64_t start_block_height);
+    void mark_last_block(uint64_t last_block_height);
+    uint64_t get_max_block_height() const;
+    void print() const;
+    std::string get_overview() const;
+    std::pair<uint64_t, uint64_t> reserve_span(uint64_t first_block_height, uint64_t last_block_height, uint64_t max_blocks, const boost::uuids::uuid &connection_id, boost::posix_time::ptime time = boost::posix_time::microsec_clock::universal_time());
+    bool is_blockchain_placeholder(const span &span) const;
+    std::pair<uint64_t, uint64_t> get_start_gap_span() const;
+    std::pair<uint64_t, uint64_t> get_next_span_if_scheduled(std::list<crypto::hash> &hashes, boost::uuids::uuid &connection_id, boost::posix_time::ptime &time) const;
+    void set_span_hashes(uint64_t start_height, const boost::uuids::uuid &connection_id, std::list<crypto::hash> hashes);
+    bool get_next_span(uint64_t &height, std::list<cryptonote::block_complete_entry> &bcel, boost::uuids::uuid &connection_id, bool filled = true) const;
+    size_t get_data_size() const;
+    size_t get_num_filled_spans_prefix() const;
+    size_t get_num_filled_spans() const;
+    crypto::hash get_last_known_hash() const;
+    float get_speed(const boost::uuids::uuid &connection_id) const;
+    bool foreach(std::function<bool(const span&)> f, bool include_blockchain_placeholder = false) const;
+
+  private:
+    block_map blocks;
+    mutable boost::recursive_mutex mutex;
+  };
+}
diff --git a/src/cryptonote_protocol/cryptonote_protocol_defs.h b/src/cryptonote_protocol/cryptonote_protocol_defs.h
index 6f6c1a803..042ae49f6 100644
--- a/src/cryptonote_protocol/cryptonote_protocol_defs.h
+++ b/src/cryptonote_protocol/cryptonote_protocol_defs.h
@@ -74,6 +74,8 @@ namespace cryptonote
   
 	uint32_t support_flags;
 
+	boost::uuids::uuid connection_id;
+
     BEGIN_KV_SERIALIZE_MAP()
       KV_SERIALIZE(incoming)
       KV_SERIALIZE(localhost)
@@ -94,6 +96,7 @@ namespace cryptonote
       KV_SERIALIZE(avg_upload)
       KV_SERIALIZE(current_upload)
       KV_SERIALIZE(support_flags)
+      KV_SERIALIZE_VAL_POD_AS_BLOB(connection_id)
     END_KV_SERIALIZE_MAP()
   };
 
diff --git a/src/cryptonote_protocol/cryptonote_protocol_handler-base.cpp b/src/cryptonote_protocol/cryptonote_protocol_handler-base.cpp
index cb4ec41af..3bda50c22 100644
--- a/src/cryptonote_protocol/cryptonote_protocol_handler-base.cpp
+++ b/src/cryptonote_protocol/cryptonote_protocol_handler-base.cpp
@@ -120,10 +120,6 @@ cryptonote_protocol_handler_base::~cryptonote_protocol_handler_base() {
 }
 
 void cryptonote_protocol_handler_base::handler_request_blocks_history(std::list<crypto::hash>& ids) {
-	using namespace epee::net_utils;
-	MDEBUG("### ~~~RRRR~~~~ ### sending request (type 2), limit = " << ids.size());
-	MDEBUG("RATE LIMIT NOT IMPLEMENTED HERE YET (download at unlimited speed?)");
-	// TODO
 }
 
 void cryptonote_protocol_handler_base::handler_response_blocks_now(size_t packet_size) {
diff --git a/src/cryptonote_protocol/cryptonote_protocol_handler.h b/src/cryptonote_protocol/cryptonote_protocol_handler.h
index 9d8bc43c2..dcee03f66 100644
--- a/src/cryptonote_protocol/cryptonote_protocol_handler.h
+++ b/src/cryptonote_protocol/cryptonote_protocol_handler.h
@@ -42,6 +42,7 @@
 #include "warnings.h"
 #include "cryptonote_protocol_defs.h"
 #include "cryptonote_protocol_handler_common.h"
+#include "block_queue.h"
 #include "cryptonote_basic/connection_context.h"
 #include "cryptonote_basic/cryptonote_stat_info.h"
 #include "cryptonote_basic/verification_context.h"
@@ -107,6 +108,7 @@ namespace cryptonote
     bool is_synchronized(){return m_synchronized;}
     void log_connections();
     std::list<connection_info> get_connections();
+    const block_queue &get_block_queue() const { return m_block_queue; }
     void stop();
   private:
     //----------------- commands handlers ----------------------------------------------
@@ -124,18 +126,19 @@ namespace cryptonote
     virtual bool relay_transactions(NOTIFY_NEW_TRANSACTIONS::request& arg, cryptonote_connection_context& exclude_context);
     //----------------------------------------------------------------------------------
     //bool get_payload_sync_data(HANDSHAKE_DATA::request& hshd, cryptonote_connection_context& context);
-    bool request_missing_objects(cryptonote_connection_context& context, bool check_having_blocks);
+    bool request_missing_objects(cryptonote_connection_context& context, bool check_having_blocks, bool force_next_span = false);
     size_t get_synchronizing_connections_count();
     bool on_connection_synchronized();
+    bool should_download_next_span(cryptonote_connection_context& context) const;
     t_core& m_core;
 
     nodetool::p2p_endpoint_stub<connection_context> m_p2p_stub;
     nodetool::i_p2p_endpoint<connection_context>* m_p2p;
     std::atomic<uint32_t> m_syncronized_connections_count;
     std::atomic<bool> m_synchronized;
-    bool m_one_request = true;
     std::atomic<bool> m_stopping;
-    epee::critical_section m_sync_lock;
+    boost::mutex m_sync_lock;
+    block_queue m_block_queue;
 
     boost::mutex m_buffer_mutex;
     double get_avg_block_size();
diff --git a/src/cryptonote_protocol/cryptonote_protocol_handler.inl b/src/cryptonote_protocol/cryptonote_protocol_handler.inl
index f3d6429a2..4652bf23e 100644
--- a/src/cryptonote_protocol/cryptonote_protocol_handler.inl
+++ b/src/cryptonote_protocol/cryptonote_protocol_handler.inl
@@ -48,6 +48,10 @@
 
 #define MLOG_P2P_MESSAGE(x) MCINFO("net.p2p.msg", context << x)
 
+#define BLOCK_QUEUE_NBLOCKS_THRESHOLD 10 // chunks of N blocks
+#define BLOCK_QUEUE_SIZE_THRESHOLD (100*1024*1024) // MB
+#define REQUEST_NEXT_SCHEDULED_SPAN_THRESHOLD (5 * 1000000) // microseconds
+
 namespace cryptonote
 {
 
@@ -233,6 +237,8 @@ namespace cryptonote
       cnx.current_download = cntxt.m_current_speed_down / 1024;
       cnx.current_upload = cntxt.m_current_speed_up / 1024;
 
+      cnx.connection_id = cntxt.m_connection_id;
+
       connections.push_back(cnx);
 
       return true;
@@ -310,6 +316,11 @@ namespace cryptonote
     MLOG_P2P_MESSAGE("Received NOTIFY_NEW_BLOCK (hop " << arg.hop << ", " << arg.b.txs.size() << " txes)");
     if(context.m_state != cryptonote_connection_context::state_normal)
       return 1;
+    if(!is_synchronized()) // can happen if a peer connection goes to normal but another thread still hasn't finished adding queued blocks
+    {
+      LOG_DEBUG_CC(context, "Received new block while syncing, ignored");
+      return 1;
+    }
     m_core.pause_mine();
     std::list<block_complete_entry> blocks;
     blocks.push_back(arg.b);
@@ -324,6 +335,7 @@ namespace cryptonote
         m_p2p->drop_connection(context);
         m_core.cleanup_handle_incoming_blocks();
         m_core.resume_mine();
+        m_block_queue.flush_spans(context.m_connection_id);
         return 1;
       }
     }
@@ -336,6 +348,7 @@ namespace cryptonote
     {
       LOG_PRINT_CCONTEXT_L0("Block verification failed, dropping connection");
       m_p2p->drop_connection(context);
+      m_block_queue.flush_spans(context.m_connection_id);
       return 1;
     }
     if(bvc.m_added_to_main_chain)
@@ -361,6 +374,11 @@ namespace cryptonote
     MLOG_P2P_MESSAGE("Received NOTIFY_NEW_FLUFFY_BLOCK (height " << arg.current_blockchain_height << ", hop " << arg.hop << ", " << arg.b.txs.size() << " txes)");
     if(context.m_state != cryptonote_connection_context::state_normal)
       return 1;
+    if(!is_synchronized()) // can happen if a peer connection goes to normal but another thread still hasn't finished adding queued blocks
+    {
+      LOG_DEBUG_CC(context, "Received new block while syncing, ignored");
+      return 1;
+    }
     
     m_core.pause_mine();
       
@@ -384,6 +402,7 @@ namespace cryptonote
           );
           
           m_p2p->drop_connection(context);
+          m_block_queue.flush_spans(context.m_connection_id);
           m_core.resume_mine();
           return 1;
         }
@@ -417,6 +436,7 @@ namespace cryptonote
               );
               
               m_p2p->drop_connection(context);
+              m_block_queue.flush_spans(context.m_connection_id);
               m_core.resume_mine();
               return 1;
             }
@@ -431,6 +451,7 @@ namespace cryptonote
             );
                         
             m_p2p->drop_connection(context);
+            m_block_queue.flush_spans(context.m_connection_id);
             m_core.resume_mine();
             return 1;
           }
@@ -455,6 +476,7 @@ namespace cryptonote
               );
               
               m_p2p->drop_connection(context);
+              m_block_queue.flush_spans(context.m_connection_id);
               m_core.resume_mine();
               return 1;
             }
@@ -472,6 +494,7 @@ namespace cryptonote
             {
               LOG_PRINT_CCONTEXT_L1("Block verification failed: transaction verification failed, dropping connection");
               m_p2p->drop_connection(context);
+              m_block_queue.flush_spans(context.m_connection_id);
               m_core.resume_mine();
               return 1;
             }
@@ -494,6 +517,7 @@ namespace cryptonote
           );
             
           m_p2p->drop_connection(context);
+          m_block_queue.flush_spans(context.m_connection_id);
           m_core.resume_mine();
           return 1;
         }
@@ -514,6 +538,7 @@ namespace cryptonote
         );
         
         m_p2p->drop_connection(context);
+        m_block_queue.flush_spans(context.m_connection_id);
         m_core.resume_mine();
         return 1;
       }      
@@ -591,6 +616,7 @@ namespace cryptonote
         {
           LOG_PRINT_CCONTEXT_L0("Block verification failed, dropping connection");
           m_p2p->drop_connection(context);
+          m_block_queue.flush_spans(context.m_connection_id);
           return 1;
         }
         if( bvc.m_added_to_main_chain )
@@ -624,6 +650,7 @@ namespace cryptonote
         
       m_core.resume_mine();
       m_p2p->drop_connection(context);
+      m_block_queue.flush_spans(context.m_connection_id);
         
       return 1;     
     }
@@ -644,6 +671,7 @@ namespace cryptonote
     {
       LOG_ERROR_CCONTEXT("failed to find block: " << arg.block_hash << ", dropping connection");
       m_p2p->drop_connection(context);
+      m_block_queue.flush_spans(context.m_connection_id);
       return 1;
     }
 
@@ -671,6 +699,7 @@ namespace cryptonote
         );
         
         m_p2p->drop_connection(context);
+        m_block_queue.flush_spans(context.m_connection_id);
         return 1;
       }
     }    
@@ -682,6 +711,7 @@ namespace cryptonote
       LOG_ERROR_CCONTEXT("Failed to handle request NOTIFY_REQUEST_FLUFFY_MISSING_TX, "
         << "failed to get requested transactions");
       m_p2p->drop_connection(context);
+      m_block_queue.flush_spans(context.m_connection_id);
       return 1;
     }
     if (!missed.empty() || txs.size() != txids.size())
@@ -689,6 +719,7 @@ namespace cryptonote
       LOG_ERROR_CCONTEXT("Failed to handle request NOTIFY_REQUEST_FLUFFY_MISSING_TX, "
         << missed.size() << " requested transactions not found" << ", dropping connection");
       m_p2p->drop_connection(context);
+      m_block_queue.flush_spans(context.m_connection_id);
       return 1;
     }
 
@@ -732,6 +763,7 @@ namespace cryptonote
       {
         LOG_PRINT_CCONTEXT_L1("Tx verification failed, dropping connection");
         m_p2p->drop_connection(context);
+        m_block_queue.flush_spans(context.m_connection_id);
         return 1;
       }
       if(tvc.m_should_be_relayed)
@@ -758,6 +790,8 @@ namespace cryptonote
     {
       LOG_ERROR_CCONTEXT("failed to handle request NOTIFY_REQUEST_GET_OBJECTS, dropping connection");
       m_p2p->drop_connection(context);
+      m_block_queue.flush_spans(context.m_connection_id);
+      return 1;
     }
     LOG_PRINT_CCONTEXT_L2("-->>NOTIFY_RESPONSE_GET_OBJECTS: blocks.size()=" << rsp.blocks.size() << ", txs.size()=" << rsp.txs.size()
                             << ", rsp.m_current_blockchain_height=" << rsp.current_blockchain_height << ", missed_ids.size()=" << rsp.missed_ids.size());
@@ -770,21 +804,15 @@ namespace cryptonote
 
 
   template<class t_core>
-  double t_cryptonote_protocol_handler<t_core>::get_avg_block_size() {
-    // return m_core.get_blockchain_storage().get_avg_block_size(count); // this does not count too well the actuall network-size of data we need to download
-
+  double t_cryptonote_protocol_handler<t_core>::get_avg_block_size()
+  {
     CRITICAL_REGION_LOCAL(m_buffer_mutex);
-    double avg = 0;
-    if (m_avg_buffer.size() == 0) {
-      _warn("m_avg_buffer.size() == 0");
+    if (m_avg_buffer.empty()) {
+      MWARNING("m_avg_buffer.size() == 0");
       return 500;
     }
-
-    const bool dbg_poke_lock = 0; // debug: try to trigger an error by poking around with locks. TODO: configure option
-    long int dbg_repeat=0;
-    do {
-      for (auto element : m_avg_buffer) avg += element;
-    } while(dbg_poke_lock && (dbg_repeat++)<100000); // in debug/poke mode, repeat this calculation to trigger hidden locking error if there is one
+    double avg = 0;
+    for (const auto &element : m_avg_buffer) avg += element;
     return avg / m_avg_buffer.size();
   }
 
@@ -794,31 +822,30 @@ namespace cryptonote
   {
     MLOG_P2P_MESSAGE("Received NOTIFY_RESPONSE_GET_OBJECTS (" << arg.blocks.size() << " blocks, " << arg.txs.size() << " txes)");
 
-    // calculate size of request - mainly for logging/debug
+    bool force_next_span = false;
+
+    // calculate size of request
     size_t size = 0;
-    for (auto element : arg.txs) size += element.size();
+    for (const auto &element : arg.txs) size += element.size();
 
-    for (auto element : arg.blocks) {
-      size += element.block.size();
-      for (auto tx : element.txs)
-        size += tx.size();
+    size_t blocks_size = 0;
+    for (const auto &element : arg.blocks) {
+      blocks_size += element.block.size();
+      for (const auto &tx : element.txs)
+        blocks_size += tx.size();
     }
+    size += blocks_size;
 
-    for (auto element : arg.missed_ids)
+    for (const auto &element : arg.missed_ids)
       size += sizeof(element.data);
 
     size += sizeof(arg.current_blockchain_height);
     {
       CRITICAL_REGION_LOCAL(m_buffer_mutex);
       m_avg_buffer.push_back(size);
-
-      const bool dbg_poke_lock = 0; // debug: try to trigger an error by poking around with locks. TODO: configure option
-      long int dbg_repeat=0;
-      do {
-        m_avg_buffer.push_back(666); // a test value
-        m_avg_buffer.erase_end(1);
-      } while(dbg_poke_lock && (dbg_repeat++)<100000); // in debug/poke mode, repeat this calculation to trigger hidden locking error if there is one
     }
+    MDEBUG(context << " downloaded " << size << " bytes worth of blocks");
+
     /*using namespace boost::chrono;
       auto point = steady_clock::now();
       auto time_from_epoh = point.time_since_epoch();
@@ -831,14 +858,17 @@ namespace cryptonote
       LOG_ERROR_CCONTEXT("sent wrong NOTIFY_HAVE_OBJECTS: arg.m_current_blockchain_height=" << arg.current_blockchain_height
         << " < m_last_response_height=" << context.m_last_response_height << ", dropping connection");
       m_p2p->drop_connection(context);
+      m_block_queue.flush_spans(context.m_connection_id);
       return 1;
     }
 
     context.m_remote_blockchain_height = arg.current_blockchain_height;
 
-    size_t count = 0;
     std::vector<crypto::hash> block_hashes;
     block_hashes.reserve(arg.blocks.size());
+    const boost::posix_time::ptime now = boost::posix_time::microsec_clock::universal_time();
+    uint64_t start_height = std::numeric_limits<uint64_t>::max();
+    cryptonote::block b;
     for(const block_complete_entry& block_entry: arg.blocks)
     {
       if (m_stopping)
@@ -846,35 +876,33 @@ namespace cryptonote
         return 1;
       }
 
-      ++count;
-      block b;
       if(!parse_and_validate_block_from_blob(block_entry.block, b))
       {
         LOG_ERROR_CCONTEXT("sent wrong block: failed to parse and validate block: "
           << epee::string_tools::buff_to_hex_nodelimer(block_entry.block) << ", dropping connection");
         m_p2p->drop_connection(context);
+        m_block_queue.flush_spans(context.m_connection_id);
         return 1;
       }
-      //to avoid concurrency in core between connections, suspend connections which delivered block later then first one
-      const crypto::hash block_hash = get_block_hash(b);
-      if(count == 2)
+      if (b.miner_tx.vin.size() != 1 || b.miner_tx.vin.front().type() != typeid(txin_gen))
       {
-        if(m_core.have_block(block_hash))
-        {
-          context.m_state = cryptonote_connection_context::state_idle;
-          context.m_needed_objects.clear();
-          context.m_requested_objects.clear();
-          LOG_PRINT_CCONTEXT_L1("Connection set to idle state.");
-          return 1;
-        }
+        LOG_ERROR_CCONTEXT("sent wrong block: block: miner tx does not have exactly one txin_gen input"
+          << epee::string_tools::buff_to_hex_nodelimer(block_entry.block) << ", dropping connection");
+        m_p2p->drop_connection(context);
+        m_block_queue.flush_spans(context.m_connection_id);
+        return 1;
       }
+      if (start_height == std::numeric_limits<uint64_t>::max())
+        start_height = boost::get<txin_gen>(b.miner_tx.vin[0]).height;
 
+      const crypto::hash block_hash = get_block_hash(b);
       auto req_it = context.m_requested_objects.find(block_hash);
       if(req_it == context.m_requested_objects.end())
       {
         LOG_ERROR_CCONTEXT("sent wrong NOTIFY_RESPONSE_GET_OBJECTS: block with id=" << epee::string_tools::pod_to_hex(get_blob_hash(block_entry.block))
           << " wasn't requested, dropping connection");
         m_p2p->drop_connection(context);
+        m_block_queue.flush_spans(context.m_connection_id);
         return 1;
       }
       if(b.tx_hashes.size() != block_entry.txs.size())
@@ -882,6 +910,7 @@ namespace cryptonote
         LOG_ERROR_CCONTEXT("sent wrong NOTIFY_RESPONSE_GET_OBJECTS: block with id=" << epee::string_tools::pod_to_hex(get_blob_hash(block_entry.block))
           << ", tx_hashes.size()=" << b.tx_hashes.size() << " mismatch with block_complete_entry.m_txs.size()=" << block_entry.txs.size() << ", dropping connection");
         m_p2p->drop_connection(context);
+        m_block_queue.flush_spans(context.m_connection_id);
         return 1;
       }
 
@@ -894,114 +923,189 @@ namespace cryptonote
       MERROR("returned not all requested objects (context.m_requested_objects.size()="
         << context.m_requested_objects.size() << "), dropping connection");
       m_p2p->drop_connection(context);
+      m_block_queue.flush_spans(context.m_connection_id);
       return 1;
     }
 
+    // get the last parsed block, which should be the highest one
+    if(m_core.have_block(cryptonote::get_block_hash(b)))
+    {
+      const uint64_t subchain_height = start_height + arg.blocks.size();
+      LOG_DEBUG_CC(context, "These are old blocks, ignoring: blocks " << start_height << " - " << (subchain_height-1) << ", blockchain height " << m_core.get_current_blockchain_height());
+      goto skip;
+    }
 
     {
       MLOG_YELLOW(el::Level::Debug, "Got NEW BLOCKS inside of " << __FUNCTION__ << ": size: " << arg.blocks.size());
 
+      // add that new span to the block queue
+      const boost::posix_time::time_duration dt = now - context.m_last_request_time;
+      const float rate = size * 1e6 / (dt.total_microseconds() + 1);
+      MDEBUG(context << " adding span: " << arg.blocks.size() << " at height " << start_height << ", " << dt.total_microseconds()/1e6 << " seconds, " << (rate/1e3) << " kB/s, size now " << (m_block_queue.get_data_size() + blocks_size) / 1048576.f << " MB");
+      m_block_queue.add_blocks(start_height, arg.blocks, context.m_connection_id, rate, blocks_size);
+
       if (m_core.get_test_drop_download() && m_core.get_test_drop_download_height()) { // DISCARD BLOCKS for testing
 
-        // we lock all the rest to avoid having multiple connections redo a lot
-        // of the same work, and one of them doing it for nothing: subsequent
-        // connections will wait until the current one's added its blocks, then
-        // will add any extra it has, if any
-        CRITICAL_REGION_LOCAL(m_sync_lock);
+        // We try to lock the sync lock. If we can, it means no other thread is
+        // currently adding blocks, so we do that for as long as we can from the
+        // block queue. Then, we go back to download.
+        const boost::unique_lock<boost::mutex> sync{m_sync_lock, boost::try_to_lock};
+        if (!sync.owns_lock())
+        {
+          MINFO("Failed to lock m_sync_lock, going back to download");
+          goto skip;
+        }
+        MDEBUG(context << " lock m_sync_lock, adding blocks to chain...");
 
         m_core.pause_mine();
         epee::misc_utils::auto_scope_leave_caller scope_exit_handler = epee::misc_utils::create_scope_leave_handler(
           boost::bind(&t_core::resume_mine, &m_core));
 
-        const uint64_t previous_height = m_core.get_current_blockchain_height();
-
-        // dismiss what another connection might already have done (likely everything)
-        uint64_t top_height;
-        crypto::hash top_hash;
-        if (m_core.get_blockchain_top(top_height, top_hash)) {
-          uint64_t dismiss = 1;
-          for (const auto &h: block_hashes) {
-            if (top_hash == h) {
-              LOG_DEBUG_CC(context, "Found current top block in synced blocks, dismissing "
-                  << dismiss << "/" << arg.blocks.size() << " blocks");
-              while (dismiss--)
-                arg.blocks.pop_front();
-              break;
-            }
-            ++dismiss;
+        while (1)
+        {
+          const uint64_t previous_height = m_core.get_current_blockchain_height();
+          uint64_t start_height;
+          std::list<cryptonote::block_complete_entry> blocks;
+          boost::uuids::uuid span_connection_id;
+          m_block_queue.mark_last_block(previous_height - 1);
+          if (!m_block_queue.get_next_span(start_height, blocks, span_connection_id))
+          {
+            MDEBUG(context << " no next span found, going back to download");
+            break;
+          }
+          MDEBUG(context << " next span in the queue has blocks " << start_height << "-" << (start_height + blocks.size() - 1)
+              << ", we need " << previous_height);
+          if (previous_height < start_height || previous_height >= start_height + blocks.size())
+          {
+            MDEBUG(context << " this span is not what we need, going back to download");
+            break;
           }
-        }
 
-        if (arg.blocks.empty())
-          goto skip;
+          const boost::posix_time::ptime start = boost::posix_time::microsec_clock::universal_time();
 
-        m_core.prepare_handle_incoming_blocks(arg.blocks);
+          m_core.prepare_handle_incoming_blocks(blocks);
 
-        for(const block_complete_entry& block_entry: arg.blocks)
-        {
-          if (m_stopping)
+          uint64_t block_process_time_full = 0, transactions_process_time_full = 0;
+          size_t num_txs = 0;
+          for(const block_complete_entry& block_entry: blocks)
           {
-              m_core.cleanup_handle_incoming_blocks();
-              return 1;
-          }
+            if (m_stopping)
+            {
+                m_core.cleanup_handle_incoming_blocks();
+                return 1;
+            }
 
-          // process transactions
-          TIME_MEASURE_START(transactions_process_time);
-          for(auto& tx_blob: block_entry.txs)
-          {
-            tx_verification_context tvc = AUTO_VAL_INIT(tvc);
-            m_core.handle_incoming_tx(tx_blob, tvc, true, true, false);
-            if(tvc.m_verifivation_failed)
+            // process transactions
+            TIME_MEASURE_START(transactions_process_time);
+            num_txs += block_entry.txs.size();
+            std::vector<tx_verification_context> tvc;
+            m_core.handle_incoming_txs(block_entry.txs, tvc, true, true, false);
+            std::list<blobdata>::const_iterator it = block_entry.txs.begin();
+            for (size_t i = 0; i < tvc.size(); ++i, ++it)
             {
-              LOG_ERROR_CCONTEXT("transaction verification failed on NOTIFY_RESPONSE_GET_OBJECTS, tx_id = "
-                  << epee::string_tools::pod_to_hex(get_blob_hash(tx_blob)) << ", dropping connection");
-              m_p2p->drop_connection(context);
+              if(tvc[i].m_verifivation_failed)
+              {
+                if (!m_p2p->for_connection(span_connection_id, [&](cryptonote_connection_context& context, nodetool::peerid_type peer_id, uint32_t f)->bool{
+                  LOG_ERROR_CCONTEXT("transaction verification failed on NOTIFY_RESPONSE_GET_OBJECTS, tx_id = "
+                      << epee::string_tools::pod_to_hex(get_blob_hash(*it)) << ", dropping connection");
+                  m_p2p->drop_connection(context);
+                  m_block_queue.flush_spans(context.m_connection_id, true);
+                  return true;
+                }))
+                  LOG_ERROR_CCONTEXT("span connection id not found");
+
+                m_core.cleanup_handle_incoming_blocks();
+                // in case the peer had dropped beforehand, remove the span anyway so other threads can wake up and get it
+                m_block_queue.remove_spans(span_connection_id, start_height);
+                return 1;
+              }
+            }
+            TIME_MEASURE_FINISH(transactions_process_time);
+            transactions_process_time_full += transactions_process_time;
+
+            // process block
+
+            TIME_MEASURE_START(block_process_time);
+            block_verification_context bvc = boost::value_initialized<block_verification_context>();
+
+            m_core.handle_incoming_block(block_entry.block, bvc, false); // <--- process block
+
+            if(bvc.m_verifivation_failed)
+            {
+              if (!m_p2p->for_connection(span_connection_id, [&](cryptonote_connection_context& context, nodetool::peerid_type peer_id, uint32_t f)->bool{
+                LOG_PRINT_CCONTEXT_L1("Block verification failed, dropping connection");
+                m_p2p->drop_connection(context);
+                m_p2p->add_host_fail(context.m_remote_address);
+                m_block_queue.flush_spans(context.m_connection_id, true);
+                return true;
+              }))
+                LOG_ERROR_CCONTEXT("span connection id not found");
+
               m_core.cleanup_handle_incoming_blocks();
+              // in case the peer had dropped beforehand, remove the span anyway so other threads can wake up and get it
+              m_block_queue.remove_spans(span_connection_id, start_height);
               return 1;
             }
-          }
-          TIME_MEASURE_FINISH(transactions_process_time);
+            if(bvc.m_marked_as_orphaned)
+            {
+              if (!m_p2p->for_connection(span_connection_id, [&](cryptonote_connection_context& context, nodetool::peerid_type peer_id, uint32_t f)->bool{
+                LOG_PRINT_CCONTEXT_L1("Block received at sync phase was marked as orphaned, dropping connection");
+                m_p2p->drop_connection(context);
+                m_p2p->add_host_fail(context.m_remote_address);
+                m_block_queue.flush_spans(context.m_connection_id, true);
+                return true;
+              }))
+                LOG_ERROR_CCONTEXT("span connection id not found");
 
-          // process block
+              m_core.cleanup_handle_incoming_blocks();
+              // in case the peer had dropped beforehand, remove the span anyway so other threads can wake up and get it
+              m_block_queue.remove_spans(span_connection_id, start_height);
+              return 1;
+            }
 
-          TIME_MEASURE_START(block_process_time);
-          block_verification_context bvc = boost::value_initialized<block_verification_context>();
+            TIME_MEASURE_FINISH(block_process_time);
+            block_process_time_full += block_process_time;
 
-          m_core.handle_incoming_block(block_entry.block, bvc, false); // <--- process block
+          } // each download block
 
-          if(bvc.m_verifivation_failed)
-          {
-            LOG_PRINT_CCONTEXT_L1("Block verification failed, dropping connection");
-            m_p2p->drop_connection(context);
-            m_p2p->add_host_fail(context.m_remote_address);
-            m_core.cleanup_handle_incoming_blocks();
-            return 1;
-          }
-          if(bvc.m_marked_as_orphaned)
-          {
-            LOG_PRINT_CCONTEXT_L1("Block received at sync phase was marked as orphaned, dropping connection");
-            m_p2p->drop_connection(context);
-            m_p2p->add_host_fail(context.m_remote_address);
-            m_core.cleanup_handle_incoming_blocks();
-            return 1;
-          }
+          MCINFO("sync-info", "Block process time (" << blocks.size() << " blocks, " << num_txs << " txs): " << block_process_time_full + transactions_process_time_full << " (" << transactions_process_time_full << "/" << block_process_time_full << ") ms");
 
-          TIME_MEASURE_FINISH(block_process_time);
-          LOG_PRINT_CCONTEXT_L2("Block process time: " << block_process_time + transactions_process_time << "(" << transactions_process_time << "/" << block_process_time << ")ms");
+          m_core.cleanup_handle_incoming_blocks();
 
-        } // each download block
-        m_core.cleanup_handle_incoming_blocks();
+          m_block_queue.mark_last_block(m_core.get_current_blockchain_height() - 1);
 
-        if (m_core.get_current_blockchain_height() > previous_height)
-        {
-          MGINFO_YELLOW(context << " Synced " << m_core.get_current_blockchain_height() << "/" << m_core.get_target_blockchain_height());
+          if (m_core.get_current_blockchain_height() > previous_height)
+          {
+            const boost::posix_time::time_duration dt = boost::posix_time::microsec_clock::universal_time() - start;
+            std::string timing_message = "";
+            if (ELPP->vRegistry()->allowed(el::Level::Info, "sync-info"))
+              timing_message = std::string(" (") + std::to_string(dt.total_microseconds()/1e6) + " sec, "
+                + std::to_string((m_core.get_current_blockchain_height() - previous_height) * 1e6 / dt.total_microseconds())
+                + " blocks/sec), " + std::to_string(m_block_queue.get_data_size() / 1048576.f) + " MB queued";
+            if (ELPP->vRegistry()->allowed(el::Level::Debug, "sync-info"))
+              timing_message += std::string(": ") + m_block_queue.get_overview();
+            MGINFO_YELLOW(context << " Synced " << m_core.get_current_blockchain_height() << "/" << m_core.get_target_blockchain_height()
+                << timing_message);
+          }
         }
       } // if not DISCARD BLOCK
 
-
+      if (should_download_next_span(context))
+      {
+        context.m_needed_objects.clear();
+        context.m_last_response_height = 0;
+        force_next_span = true;
+      }
     }
+
 skip:
-    request_missing_objects(context, true);
+    if (!request_missing_objects(context, true, force_next_span))
+    {
+      LOG_ERROR_CCONTEXT("Failed to request missing objects, dropping connection");
+      m_p2p->drop_connection(context);
+      m_block_queue.flush_spans(context.m_connection_id);
+      return 1;
+    }
     return 1;
   }
   //------------------------------------------------------------------------------------------------------------------------
@@ -1020,6 +1124,7 @@ skip:
     {
       LOG_ERROR_CCONTEXT("Failed to handle NOTIFY_REQUEST_CHAIN.");
       m_p2p->drop_connection(context);
+      m_block_queue.flush_spans(context.m_connection_id);
       return 1;
     }
     LOG_PRINT_CCONTEXT_L2("-->>NOTIFY_RESPONSE_CHAIN_ENTRY: m_start_height=" << r.start_height << ", m_total_height=" << r.total_height << ", m_block_ids.size()=" << r.m_block_ids.size());
@@ -1028,54 +1133,260 @@ skip:
   }
   //------------------------------------------------------------------------------------------------------------------------
   template<class t_core>
-  bool t_cryptonote_protocol_handler<t_core>::request_missing_objects(cryptonote_connection_context& context, bool check_having_blocks)
+  bool t_cryptonote_protocol_handler<t_core>::should_download_next_span(cryptonote_connection_context& context) const
   {
-    //if (!m_one_request == false)
-      //return true;
-    m_one_request = false;
-    // save request size to log (dr monero)
-    /*using namespace boost::chrono;
-      auto point = steady_clock::now();
-      auto time_from_epoh = point.time_since_epoch();
-      auto sec = duration_cast< seconds >( time_from_epoh ).count();*/
+    std::list<crypto::hash> hashes;
+    boost::uuids::uuid span_connection_id;
+    boost::posix_time::ptime request_time;
+    std::pair<uint64_t, uint64_t> span = m_block_queue.get_next_span_if_scheduled(hashes, span_connection_id, request_time);
 
-    if(context.m_needed_objects.size())
+    // if the next span is not scheduled (or there is none)
+    if (span.second == 0)
+    {
+      // we might be in a weird case where there is a filled next span,
+      // but it starts higher than the current height
+      uint64_t height;
+      std::list<cryptonote::block_complete_entry> bcel;
+      if (!m_block_queue.get_next_span(height, bcel, span_connection_id, true))
+        return false;
+      if (height > m_core.get_current_blockchain_height())
+      {
+        MDEBUG(context << " we should download it as the next block isn't scheduled");
+        return true;
+      }
+      return false;
+    }
+    // if it was scheduled by this particular peer
+    if (span_connection_id == context.m_connection_id)
+      return false;
+
+    float span_speed = m_block_queue.get_speed(span_connection_id);
+    float speed = m_block_queue.get_speed(context.m_connection_id);
+    MDEBUG(context << " next span is scheduled for " << span_connection_id << ", speed " << span_speed << ", ours " << speed);
+
+    // we try for that span too if:
+    //  - we're substantially faster, or:
+    //  - we're the fastest and the other one isn't (avoids a peer being waaaay slow but yet unmeasured)
+    //  - the other one asked at least 5 seconds ago
+    if (span_speed < .25 && speed > .75f)
+    {
+      MDEBUG(context << " we should download it as we're substantially faster");
+      return true;
+    }
+    if (speed == 1.0f && span_speed != 1.0f)
+    {
+      MDEBUG(context << " we should download it as we're the fastest peer");
+      return true;
+    }
+    const boost::posix_time::ptime now = boost::posix_time::microsec_clock::universal_time();
+    if ((now - request_time).total_microseconds() > REQUEST_NEXT_SCHEDULED_SPAN_THRESHOLD)
+    {
+      MDEBUG(context << " we should download it as this span was requested long ago");
+      return true;
+    }
+    return false;
+  }
+  //------------------------------------------------------------------------------------------------------------------------
+  template<class t_core>
+  bool t_cryptonote_protocol_handler<t_core>::request_missing_objects(cryptonote_connection_context& context, bool check_having_blocks, bool force_next_span)
+  {
+    m_block_queue.mark_last_block(m_core.get_current_blockchain_height() - 1);
+
+    // flush stale spans
+    std::set<boost::uuids::uuid> live_connections;
+    m_p2p->for_each_connection([&](cryptonote_connection_context& context, nodetool::peerid_type peer_id, uint32_t support_flags)->bool{
+      live_connections.insert(context.m_connection_id);
+      return true;
+    });
+    m_block_queue.flush_stale_spans(live_connections);
+
+    // if we don't need to get next span, and the block queue is full enough, wait a bit
+    bool start_from_current_chain = false;
+    if (!force_next_span)
+    {
+      bool first = true;
+      while (1)
+      {
+        size_t nblocks = m_block_queue.get_num_filled_spans();
+        size_t size = m_block_queue.get_data_size();
+        if (nblocks < BLOCK_QUEUE_NBLOCKS_THRESHOLD || size < BLOCK_QUEUE_SIZE_THRESHOLD)
+        {
+          if (!first)
+          {
+            LOG_DEBUG_CC(context, "Block queue is " << nblocks << " and " << size << ", resuming");
+          }
+          break;
+        }
+
+        if (should_download_next_span(context))
+        {
+          MDEBUG(context << " we should try for that next span too, we think we could get it faster, resuming");
+          force_next_span = true;
+          break;
+        }
+
+        if (first)
+        {
+          LOG_DEBUG_CC(context, "Block queue is " << nblocks << " and " << size << ", pausing");
+          first = false;
+        }
+        for (size_t n = 0; n < 50; ++n)
+        {
+          if (m_stopping)
+            return 1;
+          boost::this_thread::sleep_for(boost::chrono::milliseconds(100));
+        }
+      }
+    }
+
+    MDEBUG(context << " request_missing_objects: check " << check_having_blocks << ", force_next_span " << force_next_span << ", m_needed_objects " << context.m_needed_objects.size() << " lrh " << context.m_last_response_height << ", chain " << m_core.get_current_blockchain_height());
+    if(context.m_needed_objects.size() || force_next_span)
     {
       //we know objects that we need, request this objects
       NOTIFY_REQUEST_GET_OBJECTS::request req;
+      bool is_next = false;
       size_t count = 0;
-      auto it = context.m_needed_objects.begin();
-
       const size_t count_limit = m_core.get_block_sync_size();
-      MDEBUG("Setting count_limit: " << count_limit);
-      while(it != context.m_needed_objects.end() && count < count_limit)
+      std::pair<uint64_t, uint64_t> span = std::make_pair(0, 0);
+      if (force_next_span)
       {
-        if( !(check_having_blocks && m_core.have_block(*it)))
+        MDEBUG(context << " force_next_span is true, trying next span");
+        span = m_block_queue.get_start_gap_span();
+        if (span.second > 0)
         {
-          req.blocks.push_back(*it);
-          ++count;
-          context.m_requested_objects.insert(*it);
+          const uint64_t first_block_height_known = context.m_last_response_height - context.m_needed_objects.size() + 1;
+          const uint64_t last_block_height_known = context.m_last_response_height;
+          const uint64_t first_block_height_needed = span.first;
+          const uint64_t last_block_height_needed = span.first + std::min(span.second, (uint64_t)count_limit) - 1;
+          MDEBUG(context << " gap found, span: " << span.first << " - " << span.first + span.second - 1 << " (" << last_block_height_needed << ")");
+          MDEBUG(context << " current known hashes from from " << first_block_height_known << " to " << last_block_height_known);
+          if (first_block_height_needed < first_block_height_known || last_block_height_needed > last_block_height_known)
+          {
+            MDEBUG(context << " we are missing some of the necessary hashes for this gap, requesting chain again");
+            context.m_needed_objects.clear();
+            context.m_last_response_height = 0;
+            start_from_current_chain = true;
+            goto skip;
+          }
+          MDEBUG(context << " we have the hashes for this gap");
+        }
+        if (span.second == 0)
+        {
+          std::list<crypto::hash> hashes;
+          boost::uuids::uuid span_connection_id;
+          boost::posix_time::ptime time;
+          span = m_block_queue.get_next_span_if_scheduled(hashes, span_connection_id, time);
+          if (span.second > 0)
+          {
+            is_next = true;
+            for (const auto &hash: hashes)
+            {
+              req.blocks.push_back(hash);
+              context.m_requested_objects.insert(hash);
+            }
+          }
         }
-        context.m_needed_objects.erase(it++);
       }
-      LOG_PRINT_CCONTEXT_L1("-->>NOTIFY_REQUEST_GET_OBJECTS: blocks.size()=" << req.blocks.size() << ", txs.size()=" << req.txs.size()
-          << "requested blocks count=" << count << " / " << count_limit);
-      //epee::net_utils::network_throttle_manager::get_global_throttle_inreq().logger_handle_net("log/dr-monero/net/req-all.data", sec, get_avg_block_size());
+      if (span.second == 0)
+      {
+        MDEBUG(context << " span size is 0");
+        if (context.m_last_response_height + 1 < context.m_needed_objects.size())
+        {
+          MERROR(context << " ERROR: inconsistent context: lrh " << context.m_last_response_height << ", nos " << context.m_needed_objects.size());
+          context.m_needed_objects.clear();
+          context.m_last_response_height = 0;
+          goto skip;
+        }
+        const uint64_t first_block_height = context.m_last_response_height - context.m_needed_objects.size() + 1;
+        span = m_block_queue.reserve_span(first_block_height, context.m_last_response_height, count_limit, context.m_connection_id);
+        MDEBUG(context << " span from " << first_block_height << ": " << span.first << "/" << span.second);
+      }
+      if (span.second == 0 && !force_next_span)
+      {
+        MDEBUG(context << " still no span reserved, we may be in the corner case of next span scheduled and everything else scheduled/filled");
+        std::list<crypto::hash> hashes;
+        boost::uuids::uuid span_connection_id;
+        boost::posix_time::ptime time;
+        span = m_block_queue.get_next_span_if_scheduled(hashes, span_connection_id, time);
+        if (span.second > 0)
+        {
+          is_next = true;
+          for (const auto &hash: hashes)
+          {
+            req.blocks.push_back(hash);
+            context.m_requested_objects.insert(hash);
+          }
+        }
+      }
+      MDEBUG(context << " span: " << span.first << "/" << span.second << " (" << span.first << " - " << (span.first + span.second - 1) << ")");
+      if (span.second > 0)
+      {
+        if (!is_next)
+        {
+          const uint64_t first_context_block_height = context.m_last_response_height - context.m_needed_objects.size() + 1;
+          uint64_t skip = span.first - first_context_block_height;
+          if (skip > context.m_needed_objects.size())
+          {
+            MERROR("ERROR: skip " << skip << ", m_needed_objects " << context.m_needed_objects.size() << ", first_context_block_height" << first_context_block_height);
+            return false;
+          }
+          while (skip--)
+            context.m_needed_objects.pop_front();
+          if (context.m_needed_objects.size() < span.second)
+          {
+            MERROR("ERROR: span " << span.first << "/" << span.second << ", m_needed_objects " << context.m_needed_objects.size());
+            return false;
+          }
+
+          std::list<crypto::hash> hashes;
+          auto it = context.m_needed_objects.begin();
+          for (size_t n = 0; n < span.second; ++n)
+          {
+            req.blocks.push_back(*it);
+            ++count;
+            context.m_requested_objects.insert(*it);
+            hashes.push_back(*it);
+            auto j = it++;
+            context.m_needed_objects.erase(j);
+          }
+
+          m_block_queue.set_span_hashes(span.first, context.m_connection_id, hashes);
+        }
+
+        context.m_last_request_time = boost::posix_time::microsec_clock::universal_time();
+        LOG_PRINT_CCONTEXT_L1("-->>NOTIFY_REQUEST_GET_OBJECTS: blocks.size()=" << req.blocks.size() << ", txs.size()=" << req.txs.size()
+            << "requested blocks count=" << count << " / " << count_limit << " from " << span.first);
+        //epee::net_utils::network_throttle_manager::get_global_throttle_inreq().logger_handle_net("log/dr-monero/net/req-all.data", sec, get_avg_block_size());
+
+        post_notify<NOTIFY_REQUEST_GET_OBJECTS>(req, context);
+        return true;
+      }
+    }
 
-      post_notify<NOTIFY_REQUEST_GET_OBJECTS>(req, context);
-    }else if(context.m_last_response_height < context.m_remote_blockchain_height-1)
+skip:
+    context.m_needed_objects.clear();
+    if(context.m_last_response_height < context.m_remote_blockchain_height-1)
     {//we have to fetch more objects ids, request blockchain entry
 
       NOTIFY_REQUEST_CHAIN::request r = boost::value_initialized<NOTIFY_REQUEST_CHAIN::request>();
       m_core.get_short_chain_history(r.block_ids);
+
+      if (!start_from_current_chain)
+      {
+        // we'll want to start off from where we are on the download side, which may not be added yet
+        crypto::hash last_known_hash = m_block_queue.get_last_known_hash();
+        if (last_known_hash != cryptonote::null_hash && r.block_ids.front() != last_known_hash)
+          r.block_ids.push_front(last_known_hash);
+      }
+
       handler_request_blocks_history( r.block_ids ); // change the limit(?), sleep(?)
 
       //std::string blob; // for calculate size of request
       //epee::serialization::store_t_to_binary(r, blob);
       //epee::net_utils::network_throttle_manager::get_global_throttle_inreq().logger_handle_net("log/dr-monero/net/req-all.data", sec, get_avg_block_size());
-      LOG_PRINT_CCONTEXT_L1("r = " << 200);
+      //LOG_PRINT_CCONTEXT_L1("r = " << 200);
 
-      LOG_PRINT_CCONTEXT_L1("-->>NOTIFY_REQUEST_CHAIN: m_block_ids.size()=" << r.block_ids.size() );
+      LOG_PRINT_CCONTEXT_L1("-->>NOTIFY_REQUEST_CHAIN: m_block_ids.size()=" << r.block_ids.size() << ", start_from_current_chain " << start_from_current_chain);
       post_notify<NOTIFY_REQUEST_CHAIN>(r, context);
     }else
     {
@@ -1089,8 +1400,18 @@ skip:
                            << "\r\non connection [" << epee::net_utils::print_connection_context_short(context)<< "]");
 
       context.m_state = cryptonote_connection_context::state_normal;
-      MGINFO_GREEN("SYNCHRONIZED OK");
-      on_connection_synchronized();
+      if (context.m_remote_blockchain_height >= m_core.get_target_blockchain_height())
+      {
+        if (m_core.get_current_blockchain_height() >= m_core.get_target_blockchain_height())
+        {
+          MGINFO_GREEN("SYNCHRONIZED OK");
+          on_connection_synchronized();
+        }
+      }
+      else
+      {
+        MINFO(context << " we've reached this peer's blockchain height");
+      }
     }
     return true;
   }
@@ -1134,15 +1455,7 @@ skip:
       LOG_ERROR_CCONTEXT("sent empty m_block_ids, dropping connection");
       m_p2p->drop_connection(context);
       m_p2p->add_host_fail(context.m_remote_address);
-      return 1;
-    }
-
-    if(!m_core.have_block(arg.m_block_ids.front()))
-    {
-      LOG_ERROR_CCONTEXT("sent m_block_ids starting from unknown id: "
-                                              << epee::string_tools::pod_to_hex(arg.m_block_ids.front()) << " , dropping connection");
-      m_p2p->drop_connection(context);
-      m_p2p->add_host_fail(context.m_remote_address);
+      m_block_queue.flush_spans(context.m_connection_id);
       return 1;
     }
 
@@ -1154,15 +1467,22 @@ skip:
                                                                          << ", m_start_height=" << arg.start_height
                                                                          << ", m_block_ids.size()=" << arg.m_block_ids.size());
       m_p2p->drop_connection(context);
+      m_block_queue.flush_spans(context.m_connection_id);
+      return 1;
     }
 
     for(auto& bl_id: arg.m_block_ids)
     {
-      if(!m_core.have_block(bl_id))
-        context.m_needed_objects.push_back(bl_id);
+      context.m_needed_objects.push_back(bl_id);
     }
 
-    request_missing_objects(context, false);
+    if (!request_missing_objects(context, false))
+    {
+      LOG_ERROR_CCONTEXT("Failed to request missing objects, dropping connection");
+      m_p2p->drop_connection(context);
+      m_block_queue.flush_spans(context.m_connection_id);
+      return 1;
+    }
     return 1;
   }
   //------------------------------------------------------------------------------------------------------------------------