public override PerformanceRecord ReadSequential(PerfTracker perfTracker)
        {
            IEnumerable<uint> sequentialIds = Enumerable.Range(0, Constants.ReadItems).Select(x => (uint)x); ;

            return Read(string.Format("[FoundationDB] sequential read ({0} items)", Constants.ReadItems), sequentialIds,
                perfTracker);
        }

 public override List<PerformanceRecord> WriteParallelRandom(IEnumerable<TestData> data, PerfTracker perfTracker,
     int numberOfThreads, out long elapsedMilliseconds)
 {
     return
         WriteParallel(
             string.Format("[FoundationDB] parallel random write ({0} items)", Constants.ItemsPerTransaction),
             data,
             Constants.ItemsPerTransaction, Constants.WriteTransactions, perfTracker, numberOfThreads,
             out elapsedMilliseconds);
 }

        private PerformanceRecord Read(string operation, IEnumerable<uint> ids, PerfTracker perfTracker)
        {
			LightningDatabase db;
            using (var env = NewEnvironment(out db, delete: false))
            {
                var sw = Stopwatch.StartNew();

                ReadInternal(ids, perfTracker, env, db);

                sw.Stop();

                return new PerformanceRecord
                {
                    Operation = operation,
                    Time = DateTime.Now,
                    Duration = sw.ElapsedMilliseconds,
                    ProcessedItems = ids.Count()
                };
            }
        }

 public override PerformanceRecord ReadParallelRandom(IEnumerable<uint> randomIds, PerfTracker perfTracker, int numberOfThreads)
 {
     return ReadParallel(string.Format("[Voron] parallel random read ({0} items)", Constants.ReadItems), randomIds, perfTracker, numberOfThreads);
 }

        public override PerformanceRecord ReadParallelSequential(PerfTracker perfTracker, int numberOfThreads)
        {
			var sequentialIds = Enumerable.Range(0, Constants.ReadItems).Select(x => (uint)x);

            return ReadParallel(string.Format("[Voron] parallel sequential read ({0} items)", Constants.ReadItems), sequentialIds, perfTracker, numberOfThreads);
        }

        private static long ReadInternal(IEnumerable<uint> ids, PerfTracker perfTracker, StorageEnvironment env)
        {
            var ms = new byte[4096];

            using (var tx = env.NewTransaction(TransactionFlags.Read))
            {
                var sw = Stopwatch.StartNew();
                long v = 0;
                foreach (var id in ids)
                {
                    var key = id.ToString("0000000000000000");
                    var readResult = tx.State.Root.Read(tx, key);
                    int reads = 0;
                    while ((reads = readResult.Reader.Read(ms, 0, ms.Length)) > 0)
                    {
                        v += reads;
                    }
	             
                }
                perfTracker.Record(sw.ElapsedMilliseconds);
                return v;
            }
        }

        private PerformanceRecord Read(string operation, IEnumerable<uint> ids, PerfTracker perfTracker)
        {
            var options = StorageEnvironmentOptions.ForPath(dataPath);
            options.ManualFlushing = true;

            using (var env = new StorageEnvironment(options))
            {
                env.FlushLogToDataFile();

                var sw = Stopwatch.StartNew();

                var v = ReadInternal(ids, perfTracker, env);

                sw.Stop();

                return new PerformanceRecord
                {
                    Bytes = v,
                    Operation = operation,
                    Time = DateTime.Now,
                    Duration = sw.ElapsedMilliseconds,
                    ProcessedItems = ids.Count()
                };
            }
        }

        private List<PerformanceRecord> WriteInternalBatch(
            string operation,
            IEnumerator<TestData> enumerator,
            long itemsPerBatch,
            long numberOfBatches,
            PerfTracker perfTracker,
            StorageEnvironment env)
        {
            var sw = new Stopwatch();
            byte[] valueToWrite = null;
            var records = new List<PerformanceRecord>();
            for (var b = 0; b < numberOfBatches; b++)
            {
                sw.Restart();
                long v = 0;
                using (var batch = new WriteBatch())
                {
                    for (var i = 0; i < itemsPerBatch; i++)
                    {
                        enumerator.MoveNext();

                        valueToWrite = GetValueToWrite(valueToWrite, enumerator.Current.ValueSize);
                        v += valueToWrite.Length;
                        batch.Add(enumerator.Current.Id.ToString("0000000000000000"), new MemoryStream(valueToWrite), "Root");
                    }

                    env.Writer.Write(batch);
                }

                sw.Stop();
                perfTracker.Record(sw.ElapsedMilliseconds);

                records.Add(new PerformanceRecord
                {
                    Bytes = v,
                    Operation = operation,
                    Time = DateTime.Now,
                    Duration = sw.ElapsedMilliseconds,
                    ProcessedItems = itemsPerBatch
                });
            }

            return records;
        }

        private async Task<List<PerformanceRecord>> WriteInternalAsync(string operation,
            IEnumerator<TestData> enumerator,
            long itemsPerTransaction, long numberOfTransactions, PerfTracker perfTracker, FdbDatabase db)
        {
            var sw = new Stopwatch();
            byte[] valueToWrite = null;
            var records = new List<PerformanceRecord>();

            var location = db.GlobalSpace;

            sw.Restart();
            for (int transactions = 0; transactions < numberOfTransactions; transactions++)
            {
                sw.Restart();
                using (IFdbTransaction tx = db.BeginTransaction())
                {
                    for (int i = 0; i < itemsPerTransaction; i++)
                    {
                        enumerator.MoveNext();

                        valueToWrite = GetValueToWrite(valueToWrite, enumerator.Current.ValueSize);

                        tx.Set(location.Pack(enumerator.Current.Id), Slice.Create(valueToWrite));
                    }

                    await tx.CommitAsync();
                    perfTracker.Record(sw.ElapsedMilliseconds);
                }

                sw.Stop();

                records.Add(new PerformanceRecord
                {
                    Operation = operation,
                    Time = DateTime.Now,
                    Duration = sw.ElapsedMilliseconds,
                    ProcessedItems = itemsPerTransaction
                });
            }

            sw.Stop();

            return records;
        }

        private async Task<Tuple<List<PerformanceRecord>, long>> WriteParallelAsync(string operation,
            IEnumerable<TestData> data,
            int itemsPerTransaction, int numberOfTransactions, PerfTracker perfTracker, int numberOfThreads)
        {
            await NewDatabaseAsync();

            using (FdbDatabase db = await OpenDatabaseAsync())
            {
                long elapsedMilliseconds;
                List<PerformanceRecord> records = ExecuteWriteWithParallel(
                    data,
                    numberOfTransactions,
                    itemsPerTransaction,
                    numberOfThreads,
                    (enumerator, itmsPerTransaction, nmbrOfTransactions) =>
                        WriteInternal(operation, enumerator, itmsPerTransaction, nmbrOfTransactions, perfTracker, db),
                    out elapsedMilliseconds);
                return new Tuple<List<PerformanceRecord>, long>(records, elapsedMilliseconds);
            }
        }

 private List<PerformanceRecord> WriteParallel(string operation, IEnumerable<TestData> data,
     int itemsPerTransaction, int numberOfTransactions, PerfTracker perfTracker, int numberOfThreads,
     out long elapsedMilliseconds)
 {
     Tuple<List<PerformanceRecord>, long> result
         =
         WriteParallelAsync(operation, data, itemsPerTransaction, numberOfTransactions, perfTracker,
             numberOfThreads).Result;
     elapsedMilliseconds = result.Item2;
     return result.Item1;
 }

        private async Task<List<PerformanceRecord>> WriteAsync(string operation, IEnumerable<TestData> data,
            int itemsPerTransaction,
            int numberOfTransactions, PerfTracker perfTracker)
        {
            await NewDatabaseAsync();

            using (FdbDatabase db = await OpenDatabaseAsync())
            {
                IEnumerator<TestData> enumerator = data.GetEnumerator();

                return WriteInternal(operation, enumerator, itemsPerTransaction, numberOfTransactions, perfTracker, db);
            }
        }

 private List<PerformanceRecord> Write(string operation, IEnumerable<TestData> data, int itemsPerTransaction,
     int numberOfTransactions, PerfTracker perfTracker)
 {
     return WriteAsync(operation, data, itemsPerTransaction, numberOfTransactions, perfTracker).Result;
 }

        private static long ReadInternal(IEnumerable<uint> ids, PerfTracker perfTracker, LightningEnvironment env,
			LightningDatabase db)
        {
            using (var tx = env.BeginTransaction(LightningDB.TransactionBeginFlags.ReadOnly))
			using (var cursor = new LightningCursor(db, tx))
            {
                long v = 0;
                foreach (var id in ids)
                {
                    var value = cursor.MoveTo(Encoding.UTF8.GetBytes(id.ToString("0000000000000000")));
                    v += value.Value.Length;
                    //Debug.Assert(value != null);
                }
                return v;
            }
        }

        private PerformanceRecord ReadParallel(string operation, IEnumerable<uint> ids, PerfTracker perfTracker, int numberOfThreads)
        {
			LightningDatabase db;
            using (var env = NewEnvironment(out db, delete: false))
            {
                return ExecuteReadWithParallel(operation, ids, numberOfThreads, () => ReadInternal(ids, perfTracker, env, db));
            }
        }

 public override List<PerformanceRecord> WriteSequential(IEnumerable<TestData> data, PerfTracker perfTracker)
 {
     return Write(string.Format("[FoundationDB] sequential write ({0} items)", Constants.ItemsPerTransaction),
         data,
         Constants.ItemsPerTransaction, Constants.WriteTransactions, perfTracker);
 }

 private List<PerformanceRecord> WriteInternal(string operation, IEnumerator<TestData> enumerator,
     long itemsPerTransaction, long numberOfTransactions, PerfTracker perfTracker, FdbDatabase db)
 {
     return
         WriteInternalAsync(operation, enumerator, itemsPerTransaction, numberOfTransactions, perfTracker, db)
             .Result;
 }

 private PerformanceRecord Read(string operation, IEnumerable<uint> ids, PerfTracker perfTracker)
 {
     return ReadAsync(operation, ids, perfTracker).Result;
 }

        private List<PerformanceRecord> WriteParallel(string operation, IEnumerable<TestData> data, int itemsPerTransaction, int numberOfTransactions, PerfTracker perfTracker, int numberOfThreads, out long elapsedMilliseconds)
        {
            NewStorage();

            using (var env = new StorageEnvironment(StorageEnvironmentOptions.ForPath(dataPath)))
            {
                return ExecuteWriteWithParallel(
                data,
                numberOfTransactions,
                itemsPerTransaction,
                numberOfThreads,
                (enumerator, itmsPerTransaction, nmbrOfTransactions) => WriteInternalBatch(operation, enumerator, itmsPerTransaction, nmbrOfTransactions, perfTracker, env),
                out elapsedMilliseconds);
            }
        }

        private async Task<PerformanceRecord> ReadAsync(string operation, IEnumerable<uint> ids, PerfTracker perfTracker)
        {
            Stopwatch sw = Stopwatch.StartNew();

            using (FdbDatabase db = await OpenDatabaseAsync())
            {
                await ReadInternalAsync(ids, perfTracker, db);
            }

            sw.Stop();

            return new PerformanceRecord
            {
                Operation = operation,
                Time = DateTime.Now,
                Duration = sw.ElapsedMilliseconds,
                ProcessedItems = ids.Count()
            };
        }