lib/d/src/thrift/util/future.d - packaging/sources/thrift - Gitiles

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements. See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership. The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License. You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied. See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */
 module thrift.util.future;

 import core.atomic;
 import core.sync.condition;
 import core.sync.mutex;
 import core.time : Duration;
 import std.array : empty, front, popFront;
 import std.conv : to;
 import std.exception : enforce;
 import std.traits : BaseTypeTuple, isSomeFunction, ParameterTypeTuple, ReturnType;
 import thrift.base;
 import thrift.util.awaitable;
 import thrift.util.cancellation;

 /**
  * Represents an operation which is executed asynchronously and the result of
  * which will become available at some point in the future.
  *
  * Once a operation is completed, the result of the operation can be fetched
  * via the get() family of methods. There are three possible cases: Either the
  * operation succeeded, then its return value is returned, or it failed by
  * throwing, in which case the exception is rethrown, or it was cancelled
  * before, then a TCancelledException is thrown. There might be TFuture
  * implementations which never possibly enter the cancelled state.
  *
  * All methods are thread-safe, but keep in mind that any exception object or
  * result (if it is a reference type, of course) is shared between all
  * get()-family invocations.
  */
 interface TFuture(ResultType) {
   /**
    * The status the operation is currently in.
    *
    * An operation starts out in RUNNING status, and changes state to one of the
    * others at most once afterwards.
    */
   TFutureStatus status() @property;

   /**
    * A TAwaitable triggered when the operation leaves the RUNNING status.
    */
   TAwaitable completion() @property;

   /**
    * Convenience shorthand for waiting until the result is available and then
    * get()ing it.
    *
    * If the operation has already completed, the result is immediately
    * returned.
    *
    * The result of this method is »alias this«'d to the interface, so that
    * TFuture can be used as a drop-in replacement for a simple value in
    * synchronous code.
    */
   final ResultType waitGet() {
     completion.wait();
     return get();
   }
   final @property auto waitGetProperty() { return waitGet(); }
   alias waitGetProperty this;

   /**
    * Convenience shorthand for waiting until the result is available and then
    * get()ing it.
    *
    * If the operation completes in time, returns its result (resp. throws an
    * exception for the failed/cancelled cases). If not, throws a
    * TFutureException.
    */
   final ResultType waitGet(Duration timeout) {
     enforce(completion.wait(timeout), new TFutureException(
       "Operation did not complete in time."));
     return get();
   }

   /**
    * Returns the result of the operation.
    *
    * Throws: TFutureException if the operation has been cancelled,
    *   TCancelledException if it is not yet done; the set exception if it
    *   failed.
    */
   ResultType get();

   /**
    * Returns the captured exception if the operation failed, or null otherwise.
    *
    * Throws: TFutureException if not yet done, TCancelledException if the
    *   operation has been cancelled.
    */
   Exception getException();
 }

 /**
  * The states the operation offering a future interface can be in.
  */
 enum TFutureStatus : byte {
   RUNNING, /// The operation is still running.
   SUCCEEDED, /// The operation completed without throwing an exception.
   FAILED, /// The operation completed by throwing an exception.
   CANCELLED /// The operation was cancelled.
 }

 /**
  * A TFuture covering the simple but common case where the result is simply
  * set by a call to succeed()/fail().
  *
  * All methods are thread-safe, but usually, succeed()/fail() are only called
  * from a single thread (different from the thread(s) waiting for the result
  * using the TFuture interface, though).
  */
 class TPromise(ResultType) : TFuture!ResultType {
   this() {
     statusMutex_ = new Mutex;
     completionEvent_ = new TOneshotEvent;
   }

   override S status() const @property {
     return atomicLoad(status_);
   }

   override TAwaitable completion() @property {
     return completionEvent_;
   }

   override ResultType get() {
     auto s = atomicLoad(status_);
     enforce(s != S.RUNNING,
       new TFutureException("Operation not yet completed."));

     if (s == S.CANCELLED) throw new TCancelledException;
     if (s == S.FAILED) throw exception_;

     static if (!is(ResultType == void)) {
       return result_;
     }
   }

   override Exception getException() {
     auto s = atomicLoad(status_);
     enforce(s != S.RUNNING,
       new TFutureException("Operation not yet completed."));

     if (s == S.CANCELLED) throw new TCancelledException;
     if (s == S.SUCCEEDED) return null;

     return exception_;
   }

   static if (!is(ResultType == void)) {
     /**
      * Sets the result of the operation, marks it as done, and notifies any
      * waiters.
      *
      * If the operation has been cancelled before, nothing happens.
      *
      * Throws: TFutureException if the operation is already completed.
      */
     void succeed(ResultType result) {
       synchronized (statusMutex_) {
         auto s = atomicLoad(status_);
         if (s == S.CANCELLED) return;

         enforce(s == S.RUNNING,
           new TFutureException("Operation already completed."));
         result_ = result;

         atomicStore(status_, S.SUCCEEDED);
       }

       completionEvent_.trigger();
     }
   } else {
     void succeed() {
       synchronized (statusMutex_) {
         auto s = atomicLoad(status_);
         if (s == S.CANCELLED) return;

         enforce(s == S.RUNNING,
           new TFutureException("Operation already completed."));

         atomicStore(status_, S.SUCCEEDED);
       }

       completionEvent_.trigger();
     }
   }

   /**
    * Marks the operation as failed with the specified exception and notifies
    * any waiters.
    *
    * If the operation was already cancelled, nothing happens.
    *
    * Throws: TFutureException if the operation is already completed.
    */
   void fail(Exception exception) {
     synchronized (statusMutex_) {
       auto status = atomicLoad(status_);
       if (status == S.CANCELLED) return;

       enforce(status == S.RUNNING,
         new TFutureException("Operation already completed."));
       exception_ = exception;

       atomicStore(status_, S.FAILED);
     }

     completionEvent_.trigger();
   }


   /**
    * Marks this operation as completed and takes over the outcome of another
    * TFuture of the same type.
    *
    * If this operation was already cancelled, nothing happens. If the other
    * operation was cancelled, this operation is marked as failed with a
    * TCancelledException.
    *
    * Throws: TFutureException if the passed in future was not completed or
    *   this operation is already completed.
    */
   void complete(TFuture!ResultType future) {
     synchronized (statusMutex_) {
       auto status = atomicLoad(status_);
       if (status == S.CANCELLED) return;
       enforce(status == S.RUNNING,
         new TFutureException("Operation already completed."));

       enforce(future.status != S.RUNNING, new TFutureException(
         "The passed TFuture is not yet completed."));

       status = future.status;
       if (status == S.CANCELLED) {
         status = S.FAILED;
         exception_ = new TCancelledException;
       } else if (status == S.FAILED) {
         exception_ = future.getException();
       } else static if (!is(ResultType == void)) {
         result_ = future.get();
       }

       atomicStore(status_, status);
     }

     completionEvent_.trigger();
   }

   /**
    * Marks this operation as cancelled and notifies any waiters.
    *
    * If the operation is already completed, nothing happens.
    */
   void cancel() {
     synchronized (statusMutex_) {
       auto status = atomicLoad(status_);
       if (status == S.RUNNING) atomicStore(status_, S.CANCELLED);
     }

     completionEvent_.trigger();
   }

 private:
   // Convenience alias because TFutureStatus is ubiquitous in this class.
   alias TFutureStatus S;

   // The status the promise is currently in.
   shared S status_;

   union {
     static if (!is(ResultType == void)) {
       // Set if status_ is SUCCEEDED.
       ResultType result_;
     }
     // Set if status_ is FAILED.
     Exception exception_;
   }

   // Protects status_.
   // As for result_ and exception_: They are only set once, while status_ is
   // still RUNNING, so given that the operation has already completed, reading
   // them is safe without holding some kind of lock.
   Mutex statusMutex_;

   // Triggered when the event completes.
   TOneshotEvent completionEvent_;
 }

 ///
 class TFutureException : TException {
   ///
   this(string msg = "", string file = __FILE__, size_t line = __LINE__,
     Throwable next = null)
   {
     super(msg, file, line, next);
   }
 }

 /**
  * Creates an interface that is similar to a given one, but accepts an
  * additional, optional TCancellation parameter each method, and returns
  * TFutures instead of plain return values.
  *
  * For example, given the following declarations:
  * ---
  * interface Foo {
  *   void bar();
  *   string baz(int a);
  * }
  * alias TFutureInterface!Foo FutureFoo;
  * ---
  *
  * FutureFoo would be equivalent to:
  * ---
  * interface FutureFoo {
  *   TFuture!void bar(TCancellation cancellation = null);
  *   TFuture!string baz(int a, TCancellation cancellation = null);
  * }
  * ---
  */
 template TFutureInterface(Interface) if (is(Interface _ == interface)) {
   mixin({
     string code = "interface TFutureInterface \n";

     static if (is(Interface Bases == super) && Bases.length > 0) {
       code ~= ": ";
       foreach (i; 0 .. Bases.length) {
         if (i > 0) code ~= ", ";
         code ~= "TFutureInterface!(BaseTypeTuple!Interface[" ~ to!string(i) ~ "]) ";
       }
     }

     code ~= "{\n";

     foreach (methodName; __traits(derivedMembers, Interface)) {
       enum qn = "Interface." ~ methodName;
       static if (isSomeFunction!(mixin(qn))) {
         code ~= "TFuture!(ReturnType!(" ~ qn ~ ")) " ~ methodName ~
           "(ParameterTypeTuple!(" ~ qn ~ "), TCancellation cancellation = null);\n";
       }
     }

     code ~= "}\n";
     return code;
   }());
 }

 /**
  * An input range that aggregates results from multiple asynchronous operations,
  * returning them in the order they arrive.
  *
  * Additionally, a timeout can be set after which results from not yet finished
  * futures will no longer be waited for, e.g. to ensure the time it takes to
  * iterate over a set of results is limited.
  */
 final class TFutureAggregatorRange(T) {
   /**
    * Constructs a new instance.
    *
    * Params:
    *   futures = The set of futures to collect results from.
    *   timeout = If positive, not yet finished futures will be cancelled and
    *     their results will not be taken into account.
    */
   this(TFuture!T[] futures, TCancellationOrigin childCancellation,
     Duration timeout = dur!"hnsecs"(0)
   ) {
     if (timeout > dur!"hnsecs"(0)) {
       timeoutSysTick_ = TickDuration.currSystemTick +
         TickDuration.from!"hnsecs"(timeout.total!"hnsecs");
     } else {
       timeoutSysTick_ = TickDuration(0);
     }

     queueMutex_ = new Mutex;
     queueNonEmptyCondition_ = new Condition(queueMutex_);
     futures_ = futures;
     childCancellation_ = childCancellation;

     foreach (future; futures_) {
       future.completion.addCallback({
         auto f = future;
         return {
           if (f.status == TFutureStatus.CANCELLED) return;
           assert(f.status != TFutureStatus.RUNNING);

           synchronized (queueMutex_) {
             completedQueue_ ~= f;

             if (completedQueue_.length == 1) {
               queueNonEmptyCondition_.notifyAll();
             }
           }
         };
       }());
     }
   }

   /**
    * Whether the range is empty.
    *
    * This is the case if the results from the completed futures not having
    * failed have already been popped and either all future have been finished
    * or the timeout has expired.
    *
    * Potentially blocks until a new result is available or the timeout has
    * expired.
    */
   bool empty() @property {
     if (finished_) return true;
     if (bufferFilled_) return false;

     while (true) {
       TFuture!T future;
       synchronized (queueMutex_) {
         // The while loop is just being cautious about spurious wakeups, in
         // case they should be possible.
         while (completedQueue_.empty) {
           auto remaining = to!Duration(timeoutSysTick_ -
             TickDuration.currSystemTick);

           if (remaining <= dur!"hnsecs"(0)) {
             // No time left, but still no element received – we are empty now.
             finished_ = true;
             childCancellation_.trigger();
             return true;
           }

           queueNonEmptyCondition_.wait(remaining);
         }

         future = completedQueue_.front;
         completedQueue_.popFront();
       }

       ++completedCount_;
       if (completedCount_ == futures_.length) {
         // This was the last future in the list, there is no possibility
         // another result could ever become available.
         finished_ = true;
       }

       if (future.status == TFutureStatus.FAILED) {
         // This one failed, loop again and try getting another item from
         // the queue.
         exceptions_ ~= future.getException();
       } else {
         resultBuffer_ = future.get();
         bufferFilled_ = true;
         return false;
       }
     }
   }

   /**
    * Returns the first element from the range.
    *
    * Potentially blocks until a new result is available or the timeout has
    * expired.
    *
    * Throws: TException if the range is empty.
    */
   T front() {
     enforce(!empty, new TException(
       "Cannot get front of an empty future aggregator range."));
     return resultBuffer_;
   }

   /**
    * Removes the first element from the range.
    *
    * Potentially blocks until a new result is available or the timeout has
    * expired.
    *
    * Throws: TException if the range is empty.
    */
   void popFront() {
     enforce(!empty, new TException(
       "Cannot pop front of an empty future aggregator range."));
     bufferFilled_ = false;
   }

   /**
    * The number of futures the result of which has been returned or which have
    * failed so far.
    */
   size_t completedCount() @property const {
     return completedCount_;
   }

   /**
    * The exceptions collected from failed TFutures so far.
    */
   Exception[] exceptions() @property {
     return exceptions_;
   }

 private:
   TFuture!T[] futures_;
   TCancellationOrigin childCancellation_;

   // The system tick this operation will time out, or zero if no timeout has
   // been set.
   TickDuration timeoutSysTick_;

   bool finished_;

   bool bufferFilled_;
   T resultBuffer_;

   Exception[] exceptions_;
   size_t completedCount_;

   // The queue of completed futures. This (and the associated condition) are
   // the only parts of this class that are accessed by multiple threads.
   TFuture!T[] completedQueue_;
   Mutex queueMutex_;
   Condition queueNonEmptyCondition_;
 }

 /**
  * TFutureAggregatorRange construction helper to avoid having to explicitly
  * specify the value type, i.e. to allow the constructor being called using IFTI
  * (see $(DMDBUG 6082, D Bugzilla enhancement requet 6082)).
  */
 TFutureAggregatorRange!T tFutureAggregatorRange(T)(TFuture!T[] futures,
   TCancellationOrigin childCancellation, Duration timeout = dur!"hnsecs"(0)
 ) {
   return new TFutureAggregatorRange!T(futures, childCancellation, timeout);
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/
	module thrift.util.future;

	import core.atomic;
	import core.sync.condition;
	import core.sync.mutex;
	import core.time : Duration;
	import std.array : empty, front, popFront;
	import std.conv : to;
	import std.exception : enforce;
	import std.traits : BaseTypeTuple, isSomeFunction, ParameterTypeTuple, ReturnType;
	import thrift.base;
	import thrift.util.awaitable;
	import thrift.util.cancellation;

	/**
	* Represents an operation which is executed asynchronously and the result of
	* which will become available at some point in the future.
	*
	* Once a operation is completed, the result of the operation can be fetched
	* via the get() family of methods. There are three possible cases: Either the
	* operation succeeded, then its return value is returned, or it failed by
	* throwing, in which case the exception is rethrown, or it was cancelled
	* before, then a TCancelledException is thrown. There might be TFuture
	* implementations which never possibly enter the cancelled state.
	*
	* All methods are thread-safe, but keep in mind that any exception object or
	* result (if it is a reference type, of course) is shared between all
	* get()-family invocations.
	*/
	interface TFuture(ResultType) {
	/**
	* The status the operation is currently in.
	*
	* An operation starts out in RUNNING status, and changes state to one of the
	* others at most once afterwards.
	*/
	TFutureStatus status() @property;

	/**
	* A TAwaitable triggered when the operation leaves the RUNNING status.
	*/
	TAwaitable completion() @property;

	/**
	* Convenience shorthand for waiting until the result is available and then
	* get()ing it.
	*
	* If the operation has already completed, the result is immediately
	* returned.
	*
	* The result of this method is »alias this«'d to the interface, so that
	* TFuture can be used as a drop-in replacement for a simple value in
	* synchronous code.
	*/
	final ResultType waitGet() {
	completion.wait();
	return get();
	}
	final @property auto waitGetProperty() { return waitGet(); }
	alias waitGetProperty this;

	/**
	* Convenience shorthand for waiting until the result is available and then
	* get()ing it.
	*
	* If the operation completes in time, returns its result (resp. throws an
	* exception for the failed/cancelled cases). If not, throws a
	* TFutureException.
	*/
	final ResultType waitGet(Duration timeout) {
	enforce(completion.wait(timeout), new TFutureException(
	"Operation did not complete in time."));
	return get();
	}

	/**
	* Returns the result of the operation.
	*
	* Throws: TFutureException if the operation has been cancelled,
	* TCancelledException if it is not yet done; the set exception if it
	* failed.
	*/
	ResultType get();

	/**
	* Returns the captured exception if the operation failed, or null otherwise.
	*
	* Throws: TFutureException if not yet done, TCancelledException if the
	* operation has been cancelled.
	*/
	Exception getException();
	}

	/**
	* The states the operation offering a future interface can be in.
	*/
	enum TFutureStatus : byte {
	RUNNING, /// The operation is still running.
	SUCCEEDED, /// The operation completed without throwing an exception.
	FAILED, /// The operation completed by throwing an exception.
	CANCELLED /// The operation was cancelled.
	}

	/**
	* A TFuture covering the simple but common case where the result is simply
	* set by a call to succeed()/fail().
	*
	* All methods are thread-safe, but usually, succeed()/fail() are only called
	* from a single thread (different from the thread(s) waiting for the result
	* using the TFuture interface, though).
	*/
	class TPromise(ResultType) : TFuture!ResultType {
	this() {
	statusMutex_ = new Mutex;
	completionEvent_ = new TOneshotEvent;
	}

	override S status() const @property {
	return atomicLoad(status_);
	}

	override TAwaitable completion() @property {
	return completionEvent_;
	}

	override ResultType get() {
	auto s = atomicLoad(status_);
	enforce(s != S.RUNNING,
	new TFutureException("Operation not yet completed."));

	if (s == S.CANCELLED) throw new TCancelledException;
	if (s == S.FAILED) throw exception_;

	static if (!is(ResultType == void)) {
	return result_;
	}
	}

	override Exception getException() {
	auto s = atomicLoad(status_);
	enforce(s != S.RUNNING,
	new TFutureException("Operation not yet completed."));

	if (s == S.CANCELLED) throw new TCancelledException;
	if (s == S.SUCCEEDED) return null;

	return exception_;
	}

	static if (!is(ResultType == void)) {
	/**
	* Sets the result of the operation, marks it as done, and notifies any
	* waiters.
	*
	* If the operation has been cancelled before, nothing happens.
	*
	* Throws: TFutureException if the operation is already completed.
	*/
	void succeed(ResultType result) {
	synchronized (statusMutex_) {
	auto s = atomicLoad(status_);
	if (s == S.CANCELLED) return;

	enforce(s == S.RUNNING,
	new TFutureException("Operation already completed."));
	result_ = result;

	atomicStore(status_, S.SUCCEEDED);
	}

	completionEvent_.trigger();
	}
	} else {
	void succeed() {
	synchronized (statusMutex_) {
	auto s = atomicLoad(status_);
	if (s == S.CANCELLED) return;

	enforce(s == S.RUNNING,
	new TFutureException("Operation already completed."));

	atomicStore(status_, S.SUCCEEDED);
	}

	completionEvent_.trigger();
	}
	}

	/**
	* Marks the operation as failed with the specified exception and notifies
	* any waiters.
	*
	* If the operation was already cancelled, nothing happens.
	*
	* Throws: TFutureException if the operation is already completed.
	*/
	void fail(Exception exception) {
	synchronized (statusMutex_) {
	auto status = atomicLoad(status_);
	if (status == S.CANCELLED) return;

	enforce(status == S.RUNNING,
	new TFutureException("Operation already completed."));
	exception_ = exception;

	atomicStore(status_, S.FAILED);
	}

	completionEvent_.trigger();
	}


	/**
	* Marks this operation as completed and takes over the outcome of another
	* TFuture of the same type.
	*
	* If this operation was already cancelled, nothing happens. If the other
	* operation was cancelled, this operation is marked as failed with a
	* TCancelledException.
	*
	* Throws: TFutureException if the passed in future was not completed or
	* this operation is already completed.
	*/
	void complete(TFuture!ResultType future) {
	synchronized (statusMutex_) {
	auto status = atomicLoad(status_);
	if (status == S.CANCELLED) return;
	enforce(status == S.RUNNING,
	new TFutureException("Operation already completed."));

	enforce(future.status != S.RUNNING, new TFutureException(
	"The passed TFuture is not yet completed."));

	status = future.status;
	if (status == S.CANCELLED) {
	status = S.FAILED;
	exception_ = new TCancelledException;
	} else if (status == S.FAILED) {
	exception_ = future.getException();
	} else static if (!is(ResultType == void)) {
	result_ = future.get();
	}

	atomicStore(status_, status);
	}

	completionEvent_.trigger();
	}

	/**
	* Marks this operation as cancelled and notifies any waiters.
	*
	* If the operation is already completed, nothing happens.
	*/
	void cancel() {
	synchronized (statusMutex_) {
	auto status = atomicLoad(status_);
	if (status == S.RUNNING) atomicStore(status_, S.CANCELLED);
	}

	completionEvent_.trigger();
	}

	private:
	// Convenience alias because TFutureStatus is ubiquitous in this class.
	alias TFutureStatus S;

	// The status the promise is currently in.
	shared S status_;

	union {
	static if (!is(ResultType == void)) {
	// Set if status_ is SUCCEEDED.
	ResultType result_;
	}
	// Set if status_ is FAILED.
	Exception exception_;
	}

	// Protects status_.
	// As for result_ and exception_: They are only set once, while status_ is
	// still RUNNING, so given that the operation has already completed, reading
	// them is safe without holding some kind of lock.
	Mutex statusMutex_;

	// Triggered when the event completes.
	TOneshotEvent completionEvent_;
	}

	///
	class TFutureException : TException {
	///
	this(string msg = "", string file = __FILE__, size_t line = __LINE__,
	Throwable next = null)
	{
	super(msg, file, line, next);
	}
	}

	/**
	* Creates an interface that is similar to a given one, but accepts an
	* additional, optional TCancellation parameter each method, and returns
	* TFutures instead of plain return values.
	*
	* For example, given the following declarations:
	* ---
	* interface Foo {
	* void bar();
	* string baz(int a);
	* }
	* alias TFutureInterface!Foo FutureFoo;
	* ---
	*
	* FutureFoo would be equivalent to:
	* ---
	* interface FutureFoo {
	* TFuture!void bar(TCancellation cancellation = null);
	* TFuture!string baz(int a, TCancellation cancellation = null);
	* }
	* ---
	*/
	template TFutureInterface(Interface) if (is(Interface _ == interface)) {
	mixin({
	string code = "interface TFutureInterface \n";

	static if (is(Interface Bases == super) && Bases.length > 0) {
	code ~= ": ";
	foreach (i; 0 .. Bases.length) {
	if (i > 0) code ~= ", ";
	code ~= "TFutureInterface!(BaseTypeTuple!Interface[" ~ to!string(i) ~ "]) ";
	}
	}

	code ~= "{\n";

	foreach (methodName; __traits(derivedMembers, Interface)) {
	enum qn = "Interface." ~ methodName;
	static if (isSomeFunction!(mixin(qn))) {
	code ~= "TFuture!(ReturnType!(" ~ qn ~ ")) " ~ methodName ~
	"(ParameterTypeTuple!(" ~ qn ~ "), TCancellation cancellation = null);\n";
	}
	}

	code ~= "}\n";
	return code;
	}());
	}

	/**
	* An input range that aggregates results from multiple asynchronous operations,
	* returning them in the order they arrive.
	*
	* Additionally, a timeout can be set after which results from not yet finished
	* futures will no longer be waited for, e.g. to ensure the time it takes to
	* iterate over a set of results is limited.
	*/
	final class TFutureAggregatorRange(T) {
	/**
	* Constructs a new instance.
	*
	* Params:
	* futures = The set of futures to collect results from.
	* timeout = If positive, not yet finished futures will be cancelled and
	* their results will not be taken into account.
	*/
	this(TFuture!T[] futures, TCancellationOrigin childCancellation,
	Duration timeout = dur!"hnsecs"(0)
	) {
	if (timeout > dur!"hnsecs"(0)) {
	timeoutSysTick_ = TickDuration.currSystemTick +
	TickDuration.from!"hnsecs"(timeout.total!"hnsecs");
	} else {
	timeoutSysTick_ = TickDuration(0);
	}

	queueMutex_ = new Mutex;
	queueNonEmptyCondition_ = new Condition(queueMutex_);
	futures_ = futures;
	childCancellation_ = childCancellation;

	foreach (future; futures_) {
	future.completion.addCallback({
	auto f = future;
	return {
	if (f.status == TFutureStatus.CANCELLED) return;
	assert(f.status != TFutureStatus.RUNNING);

	synchronized (queueMutex_) {
	completedQueue_ ~= f;

	if (completedQueue_.length == 1) {
	queueNonEmptyCondition_.notifyAll();
	}
	}
	};
	}());
	}
	}

	/**
	* Whether the range is empty.
	*
	* This is the case if the results from the completed futures not having
	* failed have already been popped and either all future have been finished
	* or the timeout has expired.
	*
	* Potentially blocks until a new result is available or the timeout has
	* expired.
	*/
	bool empty() @property {
	if (finished_) return true;
	if (bufferFilled_) return false;

	while (true) {
	TFuture!T future;
	synchronized (queueMutex_) {
	// The while loop is just being cautious about spurious wakeups, in
	// case they should be possible.
	while (completedQueue_.empty) {
	auto remaining = to!Duration(timeoutSysTick_ -
	TickDuration.currSystemTick);

	if (remaining <= dur!"hnsecs"(0)) {
	// No time left, but still no element received – we are empty now.
	finished_ = true;
	childCancellation_.trigger();
	return true;
	}

	queueNonEmptyCondition_.wait(remaining);
	}

	future = completedQueue_.front;
	completedQueue_.popFront();
	}

	++completedCount_;
	if (completedCount_ == futures_.length) {
	// This was the last future in the list, there is no possibility
	// another result could ever become available.
	finished_ = true;
	}

	if (future.status == TFutureStatus.FAILED) {
	// This one failed, loop again and try getting another item from
	// the queue.
	exceptions_ ~= future.getException();
	} else {
	resultBuffer_ = future.get();
	bufferFilled_ = true;
	return false;
	}
	}
	}

	/**
	* Returns the first element from the range.
	*
	* Potentially blocks until a new result is available or the timeout has
	* expired.
	*
	* Throws: TException if the range is empty.
	*/
	T front() {
	enforce(!empty, new TException(
	"Cannot get front of an empty future aggregator range."));
	return resultBuffer_;
	}

	/**
	* Removes the first element from the range.
	*
	* Potentially blocks until a new result is available or the timeout has
	* expired.
	*
	* Throws: TException if the range is empty.
	*/
	void popFront() {
	enforce(!empty, new TException(
	"Cannot pop front of an empty future aggregator range."));
	bufferFilled_ = false;
	}

	/**
	* The number of futures the result of which has been returned or which have
	* failed so far.
	*/
	size_t completedCount() @property const {
	return completedCount_;
	}

	/**
	* The exceptions collected from failed TFutures so far.
	*/
	Exception[] exceptions() @property {
	return exceptions_;
	}

	private:
	TFuture!T[] futures_;
	TCancellationOrigin childCancellation_;

	// The system tick this operation will time out, or zero if no timeout has
	// been set.
	TickDuration timeoutSysTick_;

	bool finished_;

	bool bufferFilled_;
	T resultBuffer_;

	Exception[] exceptions_;
	size_t completedCount_;

	// The queue of completed futures. This (and the associated condition) are
	// the only parts of this class that are accessed by multiple threads.
	TFuture!T[] completedQueue_;
	Mutex queueMutex_;
	Condition queueNonEmptyCondition_;
	}

	/**
	* TFutureAggregatorRange construction helper to avoid having to explicitly
	* specify the value type, i.e. to allow the constructor being called using IFTI
	* (see $(DMDBUG 6082, D Bugzilla enhancement requet 6082)).
	*/
	TFutureAggregatorRange!T tFutureAggregatorRange(T)(TFuture!T[] futures,
	TCancellationOrigin childCancellation, Duration timeout = dur!"hnsecs"(0)
	) {
	return new TFutureAggregatorRange!T(futures, childCancellation, timeout);
	}