/// <summary>Base constructor for transaction filters</summary>
		/// <param name="trans">Underlying transaction that will be exposed as read-only</param>
		/// <param name="forceReadOnly">If true, force the transaction to be read-only. If false, use the read-only mode of the underlying transaction</param>
		/// <param name="ownsTransaction">If true, the underlying transaction will also be disposed when this instance is disposed</param>
		protected FdbTransactionFilter(IFdbTransaction trans, bool forceReadOnly, bool ownsTransaction)
		{
			if (trans == null) throw new ArgumentNullException("trans");

			m_transaction = trans;
			m_readOnly = forceReadOnly || trans.IsReadOnly;
			m_owner = ownsTransaction;
		}

		private async Task Scenario6(IFdbTransaction tr)
		{
			var location = FdbSubspace.Create(Slice.FromAscii("TEST"));

			tr.AtomicAdd(location.Pack("ATOMIC"), Slice.FromFixed32(0x55555555));

			var x = await tr.GetAsync(location.Pack("ATOMIC"));
			Console.WriteLine(x.ToInt32().ToString("x"));
		}

		private Task Scenario2(IFdbTransaction tr)
		{
			var location = FdbSubspace.Create(Slice.FromAscii("TEST"));
			tr.ClearRange(FdbKeyRange.StartsWith(location.Key));
			for (int i = 0; i < 10; i++)
			{
				tr.Set(location.Pack(i), Slice.FromString("value of " + i));
			}
			return Task.FromResult<object>(null);
		}

		/// <summary>Returns a 64-bit integer that
		/// 1) has never and will never be returned by another call to this
		///    method on the same subspace
		/// 2) is nearly as short as possible given the above
		/// </summary>
		public async Task<long> AllocateAsync(IFdbTransaction trans)
		{
			// find the current window size, by reading the last entry in the 'counters' subspace
			long start = 0, count = 0;
			var kv = await trans
				.Snapshot
				.GetRange(this.Counters.ToRange())
				.LastOrDefaultAsync();

			if (kv.Key.IsPresent)
			{
				start = this.Counters.UnpackSingle<long>(kv.Key);
				count = kv.Value.ToInt64();
			}

			// check if the window is full
			int window = GetWindowSize(start);
			if ((count + 1) * 2 >= window)
			{ // advance the window
				trans.ClearRange(this.Counters.Key, this.Counters.Pack(start) + FdbKey.MinValue);
				start += window;
				trans.ClearRange(this.Recent.Key, this.Recent.Pack(start));
			}

			// Increment the allocation count for the current window
			trans.AtomicAdd(this.Counters.Pack(start), Slice.FromFixed64(1));

			// As of the snapshot being read from, the window is less than half
            // full, so this should be expected to take 2 tries.  Under high
            // contention (and when the window advances), there is an additional
            // subsequent risk of conflict for this transaction.
			while (true)
			{
				// Find a random free slot in the current window...
				long candidate;
				lock (m_rnd)
				{
					candidate = start + m_rnd.Next(window);
				}

				// test if the key is used
				var key = this.Recent.Pack(candidate);
				var value = await trans.GetAsync(key).ConfigureAwait(false);

				if (value.IsNull)
				{ // free slot

					// mark as used
					trans.Set(key, Slice.Empty);
					return candidate;
				}

				// no luck this time, try again...
			}
		}

		private Task Scenario4(IFdbTransaction tr)
		{
			var location = FdbSubspace.Create(Slice.FromAscii("TEST"));

			//tr.Set(location.Key, Slice.FromString("NARF"));
			//tr.AtomicAdd(location.Key, Slice.FromFixedU32(1));
			tr.AtomicAnd(location.Key, Slice.FromFixedU32(7));
			tr.AtomicXor(location.Key, Slice.FromFixedU32(3));
			tr.AtomicXor(location.Key, Slice.FromFixedU32(15));
			return Task.FromResult<object>(null);
		}

		private Task Scenario3(IFdbTransaction tr)
		{
			var location = FdbSubspace.Create(Slice.FromAscii("TEST"));

			tr.Set(location.Key + (byte)'a', Slice.FromAscii("A"));
			tr.AtomicAdd(location.Key + (byte)'k', Slice.FromFixed32(1));
			tr.Set(location.Key + (byte)'z', Slice.FromAscii("C"));
			tr.ClearRange(location.Key + (byte)'a', location.Key + (byte)'k');
			tr.ClearRange(location.Key + (byte)'k', location.Key + (byte)'z');
			return Task.FromResult<object>(null);
		}

		private async Task Scenario5(IFdbTransaction tr)
		{
			var location = FdbSubspace.Create(Slice.FromAscii("TEST"));

			//tr.Set(location.Pack(42), Slice.FromString("42"));
			//tr.Set(location.Pack(50), Slice.FromString("50"));
			//tr.Set(location.Pack(60), Slice.FromString("60"));

			var x = await tr.GetKeyAsync(FdbKeySelector.LastLessThan(location.Pack(49)));
			Console.WriteLine(x);

			tr.Set(location.Pack("FOO"), Slice.FromString("BAR"));

		}

		/// <summary>Remove all fields of an hashset</summary>
		/// <param name="id"></param>
		public void Delete(IFdbTransaction trans, IFdbTuple id)
		{
			if (trans == null) throw new ArgumentNullException("trans");
			if (id == null) throw new ArgumentNullException("id");

			// remove all fields of the hash
			trans.ClearRange(FdbKeyRange.StartsWith(GetKey(id)));
		}

		public void Set(IFdbTransaction trans, IFdbTuple id, IEnumerable<KeyValuePair<string, Slice>> fields)
		{
			if (trans == null) throw new ArgumentNullException("trans");
			if (id == null) throw new ArgumentNullException("id");
			if (fields == null) throw new ArgumentNullException("fields");

			foreach (var field in fields)
			{
				if (string.IsNullOrEmpty(field.Key)) throw new ArgumentException("Field cannot have an empty name", "fields");
				trans.Set(GetFieldKey(id, field.Key), field.Value);
			}
		}

		private async Task<Slice> GetPreviousNodeAsync(IFdbTransaction trans, int level, Slice key)
		{
			// GetPreviousNodeAsync looks for the previous node on a level, but "doesn't care"
			// about the contents of that node. It therefore uses a non-isolated (snaphot)
			// read and explicitly adds a conflict range that is exclusive of the actual,
			// found previous node. This allows an increment of that node not to trigger
			// a transaction conflict. We also add a conflict key on the found previous
			// key in level 0. This allows detection of erasures.

			var k = this.Subspace.Pack(level, key);
			//Console.WriteLine(k);
			//Console.WriteLine("GetPreviousNode(" + level + ", " + key + ")");
			//Console.WriteLine(FdbKeySelector.LastLessThan(k) + " <= x < " + FdbKeySelector.FirstGreaterOrEqual(k));
			var kv = await trans
				.Snapshot
				.GetRange(
					FdbKeySelector.LastLessThan(k),
					FdbKeySelector.FirstGreaterOrEqual(k)
				)
				.FirstAsync()
				.ConfigureAwait(false);
			//Console.WriteLine("Found " + FdbKey.Dump(kv.Key));

			var prevKey = this.Subspace.UnpackLast<Slice>(kv.Key);
			trans.AddReadConflictRange(kv.Key + FdbKey.MinValue, k);
			trans.AddReadConflictKey(this.Subspace.Pack(0, prevKey));
			return prevKey;
		}

		private void ClearTask(IFdbTransaction tr, Slice taskId)
		{
			tr.Annotate("Deleting task {0}", taskId.ToAsciiOrHexaString());

			// clear all metadata about the task
			tr.ClearRange(FdbKeyRange.StartsWith(this.TaskStore.Pack(taskId)));
			// decrement pending number of tasks
			this.Counters.Decrement(tr, COUNTER_PENDING_TASKS);
		}

		private async Task PushQueueAsync(IFdbTransaction tr, FdbSubspace queue, Slice taskId)
		{
			//TODO: use a high contention algo ?
			// - must support Push and Pop
			// - an empty queue must correspond to an empty subspace

			// get the current size of the queue
			var range = queue.ToRange();
			var lastKey = await tr.Snapshot.GetKeyAsync(FdbKeySelector.LastLessThan(range.End)).ConfigureAwait(false);
			int count = lastKey < range.Begin ? 0 : queue.Unpack(lastKey).Get<int>(0) + 1;

			// set the value
			tr.Set(queue.Pack(count, GetRandomId()), taskId);
		}

		public PrefixRewriterTransaction(FdbSubspace prefix, IFdbTransaction trans, bool ownsTransaction)
			: base(trans, false, ownsTransaction)
		{
			if (prefix == null) throw new ArgumentNullException("prefix");
			m_prefix = prefix;
		}

		/// <summary>Marks the end of the transaction</summary>
		/// <param name="trans"></param>
		public void Stop(IFdbTransaction trans)
		{
			if (!this.Completed)
			{
				this.Completed = true;
				this.StopTimestamp = GetTimestamp();
				this.StoppedUtc = DateTimeOffset.UtcNow;
			}
		}

		/// <summary>Marks the start of the transaction</summary>
		/// <param name="trans"></param>
		public void Start(IFdbTransaction trans)
		{
			this.Id = trans.Id;
			this.StartedUtc = DateTimeOffset.UtcNow;
			this.StartTimestamp = GetTimestamp();
		}

		public Task<FdbDirectorySubspace> TryMoveToAsync(IFdbTransaction trans, IEnumerable<string> newAbsolutePath)
		{
			throw new NotSupportedException("Database partitions cannot be moved");
		}

		Task<bool> IFdbDirectory.TryRemoveAsync(IFdbTransaction trans, IEnumerable<string> path)
		{
			return m_directory.TryRemoveAsync(trans, path);
		}

		private async Task<KeyValuePair<Slice, Slice>> FindRandomItem(IFdbTransaction tr, FdbSubspace ring)
		{
			var range = ring.ToRange();

			// start from a random position around the ring
			Slice key = ring.Pack(GetRandomId());

			// We want to find the next item in the clockwise direction. If we reach the end of the ring, we "wrap around" by starting again from the start
			// => So we do find_next(key <= x < MAX) and if that does not produce any result, we do a find_next(MIN <= x < key)

			// When the ring only contains a few items (or is empty), there is more than 50% change that we wont find anything in the first read.
			// To reduce the latency for this case, we will issue both range reads at the same time, and discard the second one if the first returned something.
			// This should reduce the latency in half when the ring is empty, or when it contains only items before the random key.

			var candidate = await tr.GetRange(key, range.End).FirstOrDefaultAsync();

			if (!candidate.Key.IsPresent)
			{
				candidate = await tr.GetRange(range.Begin, key).FirstOrDefaultAsync();
			}

			return candidate;
		}

		/// <summary>Attempts to move the specified subdirectory to <paramref name="newPath"/>.
		/// There is no effect on the physical prefix of the given directory, or on clients that already have the directory open.
		/// An error is raised if a directory already exists at `new_path`.
		/// </summary>
		/// <param name="trans">Transaction to use for the operation</param>
		/// <param name="oldPath">Relative path under this directory of the subdirectory to be moved</param>
		/// <param name="newPath">Relative path under this directory where the subdirectory will be moved to</param>
		/// <returns>Returns the directory at its new location if successful. If the directory cannot be moved, then null is returned.</returns>
		Task<FdbDirectorySubspace> IFdbDirectory.TryMoveAsync(IFdbTransaction trans, IEnumerable<string> oldPath, IEnumerable<string> newPath)
		{
			return this.DirectoryLayer.TryMoveAsync(trans, this.ToRelativePath(oldPath).ToArray<string>(), this.ToRelativePath(newPath).ToArray<string>());
		}

		private void StoreTask(IFdbTransaction tr, Slice taskId, DateTime scheduledUtc, Slice taskBody)
		{
			tr.Annotate("Writing task {0}", taskId.ToAsciiOrHexaString());

			var prefix = this.TaskStore.Partition(taskId);

			// store task body and timestamp
			tr.Set(prefix.Key, taskBody);
			tr.Set(prefix.Pack(TASK_META_SCHEDULED), Slice.FromInt64(scheduledUtc.Ticks));
			// increment total and pending number of tasks
			this.Counters.Increment(tr, COUNTER_TOTAL_TASKS);
			this.Counters.Increment(tr, COUNTER_PENDING_TASKS);
		}