Queue all zonefile hashes in the BlockstackDB
    to the zonefile queue

    NOT THREAD SAFE

    Returns the list of zonefile infos queued, and whether or not they are present.
544265
def atlasdb_queue_zonefiles( con, db, start_block, zonefile_dir, recover=False, validate=True, end_block=None ):
    
    # populate zonefile queue
    total = 0
    if end_block is None:
        end_block = db.lastblock+1

    ret = []    # map zonefile hash to zfinfo

    for block_height in range(start_block, end_block, 1):
        
        # TODO: can we do this transactionally?
        zonefile_info = db.get_atlas_zonefile_info_at( block_height )
        for name_txid_zfhash in zonefile_info:
            name = str(name_txid_zfhash['name'])
            zfhash = str(name_txid_zfhash['value_hash'])
            txid = str(name_txid_zfhash['txid'])
            tried_storage = 0

            present = is_zonefile_cached( zfhash, zonefile_dir, validate=validate )
            zfinfo = atlasdb_get_zonefile( zfhash, con=con )
            if zfinfo is not None:
                tried_storage = zfinfo['tried_storage']

            if recover and present:
                log.debug('Recover: assume that {} is absent so we will reprocess it'.format(zfhash))
                present = False

            log.debug("Add %s %s %s at %s (present: %s, tried_storage: %s)" % (name, zfhash, txid, block_height, present, tried_storage) )
            atlasdb_add_zonefile_info( name, zfhash, txid, present, tried_storage, block_height, con=con )
            total += 1

            ret.append({
                'name': name,
                'zonefile_hash': zfhash,
                'txid': txid,
                'block_height': block_height,
                'present': present,
                'tried_storage': tried_storage
            })

    log.debug("Queued %s zonefiles from %s-%s" % (total, start_block, db.lastblock))
    return ret
1271124
def is_zonefile_cached( zonefile_hash, zonefile_dir, validate=False):
    
    zonefile_path = atlas_zonefile_path(zonefile_dir, zonefile_hash)
    zonefile_path_legacy = atlas_zonefile_path_legacy(zonefile_dir, zonefile_hash)

    res = False
    for zfp in [zonefile_path, zonefile_path_legacy]:
        
        if not os.path.exists(zfp):
            continue

        if validate:
            data = _read_atlas_zonefile(zfp, zonefile_hash)
            if data:
                # yup!
                res = True
                break

        else:
            res = True
            break

    return res
717340
def get_block_info(self):
        
        if not self.finished:
            raise Exception("Not finished downloading")

        ret = []
        for (block_hash, block_data) in self.block_info.items():
            ret.append( (block_data['height'], block_data['txns']) )

        return ret
902256
public void lockBlocks(long fileId, List<Long> blockIds) throws IOException {
    Map<Long, Long> blockIdToLockId = new HashMap<>();
    List<Throwable> errors = new ArrayList<>();
    synchronized (mLock) {
      if (mPersistingInProgressFiles.containsKey(fileId)) {
        throw new IOException("the file " + fileId + " is already being persisted");
      }
    }
    try {
      // lock all the blocks to prevent any eviction
      for (long blockId : blockIds) {
        long lockId = mBlockWorker.lockBlock(Sessions.CHECKPOINT_SESSION_ID, blockId);
        blockIdToLockId.put(blockId, lockId);
      }
    } catch (BlockDoesNotExistException e) {
      errors.add(e);
      // make sure all the locks are released
      for (long lockId : blockIdToLockId.values()) {
        try {
          mBlockWorker.unlockBlock(lockId);
        } catch (BlockDoesNotExistException bdnee) {
          errors.add(bdnee);
        }
      }

      if (!errors.isEmpty()) {
        StringBuilder errorStr = new StringBuilder();
        errorStr.append("failed to lock all blocks of file ").append(fileId).append("\n");
        for (Throwable error : errors) {
          errorStr.append(error).append('\n');
        }
        throw new IOException(errorStr.toString());
      }
    }
    synchronized (mLock) {
      mPersistingInProgressFiles.put(fileId, blockIdToLockId);
    }
  }
726895
def flush(self, safe=None):
		''' '''
		result = None
		for index, op in enumerate(self.queue):
			try:
				result = op.execute()
			except:
				self.clear_queue()
				self.clear_cache()
				raise
		self.clear_queue()
		return result
955810
def db_continue( block_id, consensus_hash ):
    

    # every so often, clean up
    if (block_id % 20) == 0:
        log.debug("Pre-emptive garbage collection at %s" % block_id)
        gc.collect(2)

    return is_running() or os.environ.get("BLOCKSTACK_TEST") == "1"
562795
def is_subdomain_zonefile_hash(self, fqn, zonefile_hash, cur=None):
        
        sql = 'SELECT COUNT(zonefile_hash) FROM {} WHERE fully_qualified_subdomain = ? and zonefile_hash = ?;'.format(self.subdomain_table)
        args = (fqn,zonefile_hash)

        cursor = None
        if cur is None:
            cursor = self.conn.cursor()
        else:
            cursor = cur

        rows = db_query_execute(cursor, sql, args)
        
        count = None
        for row in rows:
            count = row['COUNT(zonefile_hash)']
            break

        return (count > 0)
328158
private File getWriteLock() {
        File[] lckFiles = this.lockPath.listFiles(lockFilter);
        for (final File lckFile : lckFiles) {
            if (lockFilter.isWriterLockFile(lckFile)) {
                return lckFile;
            }
        }

        return null;
    }
549410
function() {

        var fileIndex;

        if (stop_loop) {
          return false;
        }

        // Check to see if are in queue mode
        if (opts.queuefiles > 0 && processingQueue.length >= opts.queuefiles) {
          return pause(opts.queuewait);
        } else {
          // Take first thing off work queue
          fileIndex = workQueue[0];
          workQueue.splice(0, 1);

          // Add to processing queue
          processingQueue.push(fileIndex);
        }

        try {
          if (beforeEach(files[fileIndex]) !== false) {
            if (fileIndex === files_count) {
              return;
            }
            var reader = new FileReader(),
                max_file_size = 1048576 * opts.maxfilesize;

            reader.index = fileIndex;
            if (files[fileIndex].size > max_file_size) {
              opts.error(errors[2], files[fileIndex], fileIndex);
              // Remove from queue
              processingQueue.forEach(function(value, key) {
                if (value === fileIndex) {
                  processingQueue.splice(key, 1);
                }
              });
              filesRejected++;
              return true;
            }

            reader.onloadend = !opts.beforeSend ? send : function (e) {
              opts.beforeSend(files[fileIndex], fileIndex, function () { send(e); });
            };

            reader.readAsBinaryString(files[fileIndex]);

          } else {
            filesRejected++;
          }
        } catch (err) {
          // Remove from queue
          processingQueue.forEach(function(value, key) {
            if (value === fileIndex) {
              processingQueue.splice(key, 1);
            }
          });
          opts.error(errors[0]);
          return false;
        }

        // If we still have work to do,
        if (workQueue.length > 0) {
          process();
        }
      }
658624
def perform
      unless @threads.any? { |t| t.alive? }
        @values.start!
        @size.times do
          @threads << Thread.new do
            while job = @queue.pop
              if job == :END_OF_WORK
                break
              else
                @values.push *job[1], &job[0]
              end
            end
          end
        end
      end
    end
871327
def GetHashData(hashable):
        
        ms = StreamManager.GetStream()
        writer = BinaryWriter(ms)
        hashable.SerializeUnsigned(writer)
        ms.flush()
        retVal = ms.ToArray()
        StreamManager.ReleaseStream(ms)
        return retVal
278027
def get_consensus_hashes(cls, impl, working_dir, start_block_height=None, end_block_height=None, db_con=None, completeness_check=True):
        
        if (start_block_height is None and end_block_height is not None) or (start_block_height is not None and end_block_height is None):
            raise ValueError("Need either both or neither start/end block height")

        we_opened = False
        if db_con is not None:
            con = db_con

        else:
            assert impl and working_dir, 'Need impl and working_dir if db_con is not given'
            con = cls.db_open(impl, working_dir)
            we_opened = True

        range_query = ''
        range_args = ()

        if start_block_height and end_block_height:
            range_query += ' WHERE block_id >= ? AND block_id < ?'
            range_args += (start_block_height,end_block_height)

        query = 'SELECT block_id,consensus_hash FROM snapshots' + range_query + ';'
        args = range_args

        rows = cls.db_query_execute(con, query, range_args, verbose=False)
        ret = {}
        block_min = None
        block_max = None

        for r in rows:
            block_id = int(r['block_id'])
            ret[block_id] = r['consensus_hash']

            if block_min is None or block_min > block_id:
                block_min = block_id

            if block_max is None or block_max < block_id:
                block_max = block_id

        if we_opened:
            con.close()
       
        if completeness_check:
            # sanity check
            for i in range(block_min,block_max+1):
                ch = ret.get(i, None)

                assert ch is not None, 'Missing consensus hash for {}'.format(i)
                assert isinstance(ret[i], (str,unicode)), 'consensus hash for {} is type {}'.format(i, type(ret[i]))

        return ret
510764
def atlasdb_add_zonefile_info( name, zonefile_hash, txid, present, tried_storage, block_height, con=None, path=None ):
    
    global ZONEFILE_INV, NUM_ZONEFILES, ZONEFILE_INV_LOCK

    with AtlasDBOpen( con=con, path=path ) as dbcon:
        with ZONEFILE_INV_LOCK:
            # need to lock here since someone could call atlasdb_cache_zonefile_info
            if present:
                present = 1
            else:
                present = 0

            if tried_storage:
                tried_storage = 1
            else:
                tried_storage = 0

            sql = "UPDATE zonefiles SET name = ?, zonefile_hash = ?, txid = ?, present = ?, tried_storage = ?, block_height = ? WHERE txid = ?;"
            args = (name, zonefile_hash, txid, present, tried_storage, block_height, txid )

            cur = dbcon.cursor()
            update_res = atlasdb_query_execute( cur, sql, args )
            dbcon.commit()

            if update_res.rowcount == 0:
                sql = "INSERT OR IGNORE INTO zonefiles (name, zonefile_hash, txid, present, tried_storage, block_height) VALUES (?,?,?,?,?,?);"
                args = (name, zonefile_hash, txid, present, tried_storage, block_height)
            
                cur = dbcon.cursor()
                atlasdb_query_execute( cur, sql, args )
                dbcon.commit()

            # keep in-RAM zonefile inv coherent
            zfbits = atlasdb_get_zonefile_bits( zonefile_hash, con=dbcon, path=path )

            inv_vec = None
            if ZONEFILE_INV is None:
                inv_vec = ""
            else:
                inv_vec = ZONEFILE_INV[:]

            ZONEFILE_INV = atlas_inventory_flip_zonefile_bits( inv_vec, zfbits, present )
            log.debug('Set {} ({}) to {}'.format(zonefile_hash, ','.join(str(i) for i in zfbits), present))

            # keep in-RAM zonefile count coherent
            NUM_ZONEFILES = atlasdb_zonefile_inv_length( con=dbcon, path=path )

    return True
1103378
def _read_atlas_zonefile( zonefile_path, zonefile_hash ):
    

    with open(zonefile_path, "rb") as f:
        data = f.read()

    # sanity check 
    if zonefile_hash is not None:
        if not verify_zonefile( data, zonefile_hash ):
            log.debug("Corrupt zonefile '%s'" % zonefile_hash)
            return None

    return data
715729
def atlasdb_cache_zonefile_info( con=None, path=None ):
    
    global ZONEFILE_INV, NUM_ZONEFILES, ZONEFILE_INV_LOCK
    
    inv = None
    with ZONEFILE_INV_LOCK:
        inv_len = atlasdb_zonefile_inv_length( con=con, path=path )
        inv = atlas_make_zonefile_inventory( 0, inv_len, con=con, path=path )

        ZONEFILE_INV = inv
        NUM_ZONEFILES = inv_len

    return inv
1266458
HashSet getQueuePointGuessSet()
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "getQueuePointGuessSet");

    HashSet theQueuePoints = null;

    synchronized (_queuePointsGuessSet)
    {
      theQueuePoints = (HashSet) _queuePointsGuessSet.clone();
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "getQueuePointGuessSet", theQueuePoints);

    return theQueuePoints;
  }
236763
private synchronized List<TableBucket> toBuckets(List<Map.Entry<UUID, Long>> indexHashes) {
        val buckets = new ArrayList<TableBucket>();
        if (indexHashes == null) {
            // Nothing more in the index. Add whatever is in the cache attributes.
            while (!this.cacheHashes.isEmpty()) {
                add(this.cacheHashes.removeFirst(), buckets);
            }

            return buckets.isEmpty() ? null : buckets;
        } else {
            // Transform every eligible Attribute into a TableBucket and add it to the result.
            for (val indexHash : indexHashes) {
                if (KeyHasher.isValid(indexHash.getKey()) && indexHash.getValue() != Attributes.NULL_ATTRIBUTE_VALUE) {
                    // For each bucket returned above, include all Buckets/hashes from the ContainerKeyIndex which are equal to or
                    // below it. (this is very similar to the AttributeMixer - maybe reuse that methodology).
                    boolean overridden = false;
                    while (!this.cacheHashes.isEmpty()) {
                        val cacheHash = this.cacheHashes.peekFirst();
                        int cmp = indexHash.getKey().compareTo(this.cacheHashes.peekFirst().getKey());
                        if (cmp < 0) {
                            // Cache Hash is after Index Hash. We are done here.
                            break;
                        }

                        // The value we got from the Index was overridden by the one in the cache.
                        overridden = overridden || cmp == 0;
                        add(cacheHash, buckets);
                        this.cacheHashes.removeFirst();
                    }

                    if (!overridden) {
                        add(indexHash, buckets);
                    }
                }
            }
        }

        return buckets;
    }