--- /dev/null
+/*
+ * 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.
+ */
+
+/**
+ * Utilities for asynchronously querying multiple servers, building on
+ * TAsyncClient.
+ *
+ * Terminology note: The names of the artifacts defined in this module are
+ * derived from »client pool«, because they operate on a pool of
+ * TAsyncClients. However, from a architectural point of view, they often
+ * represent a pool of hosts a Thrift client application communicates with
+ * using RPC calls.
+ */
+module thrift.codegen.async_client_pool;
+
+import core.sync.mutex;
+import core.time : Duration, dur;
+import std.algorithm : map;
+import std.array : array, empty;
+import std.exception : enforce;
+import std.traits : ParameterTypeTuple, ReturnType;
+import thrift.base;
+import thrift.codegen.base;
+import thrift.codegen.async_client;
+import thrift.internal.algorithm;
+import thrift.internal.codegen;
+import thrift.util.awaitable;
+import thrift.util.cancellation;
+import thrift.util.future;
+import thrift.internal.resource_pool;
+
+/**
+ * Represents a generic client pool which implements TFutureInterface!Interface
+ * using multiple TAsyncClients.
+ */
+interface TAsyncClientPoolBase(Interface) if (isService!Interface) :
+ TFutureInterface!Interface
+{
+ /// Shorthand for the client type this pool operates on.
+ alias TAsyncClientBase!Interface Client;
+
+ /**
+ * Adds a client to the pool.
+ */
+ void addClient(Client client);
+
+ /**
+ * Removes a client from the pool.
+ *
+ * Returns: Whether the client was found in the pool.
+ */
+ bool removeClient(Client client);
+
+ /**
+ * Called to determine whether an exception comes from a client from the
+ * pool not working properly, or if it an exception thrown at the
+ * application level.
+ *
+ * If the delegate returns true, the server/connection is considered to be
+ * at fault, if it returns false, the exception is just passed on to the
+ * caller.
+ *
+ * By default, returns true for instances of TTransportException and
+ * TApplicationException, false otherwise.
+ */
+ bool delegate(Exception) rpcFaultFilter() const @property;
+ void rpcFaultFilter(bool delegate(Exception)) @property; /// Ditto
+
+ /**
+ * Whether to open the underlying transports of a client before trying to
+ * execute a method if they are not open. This is usually desirable
+ * because it allows e.g. to automatically reconnect to a remote server
+ * if the network connection is dropped.
+ *
+ * Defaults to true.
+ */
+ bool reopenTransports() const @property;
+ void reopenTransports(bool) @property; /// Ditto
+}
+
+immutable bool delegate(Exception) defaultRpcFaultFilter;
+static this() {
+ defaultRpcFaultFilter = (Exception e) {
+ import thrift.protocol.base;
+ import thrift.transport.base;
+ return (
+ (cast(TTransportException)e !is null) ||
+ (cast(TApplicationException)e !is null)
+ );
+ };
+}
+
+/**
+ * A TAsyncClientPoolBase implementation which queries multiple servers in a
+ * row until a request succeeds, the result of which is then returned.
+ *
+ * The definition of »success« can be customized using the rpcFaultFilter()
+ * delegate property. If it is non-null and calling it for an exception set by
+ * a failed method invocation returns true, the error is considered to be
+ * caused by the RPC layer rather than the application layer, and the next
+ * server in the pool is tried. If there are no more clients to try, the
+ * operation is marked as failed with a TCompoundOperationException.
+ *
+ * If a TAsyncClient in the pool fails with an RPC exception for a number of
+ * consecutive tries, it is temporarily disabled (not tried any longer) for
+ * a certain duration. Both the limit and the timeout can be configured. If all
+ * clients fail (and keepTrying is false), the operation fails with a
+ * TCompoundOperationException which contains the collected RPC exceptions.
+ */
+final class TAsyncClientPool(Interface) if (isService!Interface) :
+ TAsyncClientPoolBase!Interface
+{
+ ///
+ this(Client[] clients) {
+ pool_ = new TResourcePool!Client(clients);
+ rpcFaultFilter_ = defaultRpcFaultFilter;
+ reopenTransports_ = true;
+ }
+
+ /+override+/ void addClient(Client client) {
+ pool_.add(client);
+ }
+
+ /+override+/ bool removeClient(Client client) {
+ return pool_.remove(client);
+ }
+
+ /**
+ * Whether to keep trying to find a working client if all have failed in a
+ * row.
+ *
+ * Defaults to false.
+ */
+ bool keepTrying() const @property {
+ return pool_.cycle;
+ }
+
+ /// Ditto
+ void keepTrying(bool value) @property {
+ pool_.cycle = value;
+ }
+
+ /**
+ * Whether to use a random permutation of the client pool on every call to
+ * execute(). This can be used e.g. as a simple form of load balancing.
+ *
+ * Defaults to true.
+ */
+ bool permuteClients() const @property {
+ return pool_.permute;
+ }
+
+ /// Ditto
+ void permuteClients(bool value) @property {
+ pool_.permute = value;
+ }
+
+ /**
+ * The number of consecutive faults after which a client is disabled until
+ * faultDisableDuration has passed. 0 to never disable clients.
+ *
+ * Defaults to 0.
+ */
+ ushort faultDisableCount() const @property {
+ return pool_.faultDisableCount;
+ }
+
+ /// Ditto
+ void faultDisableCount(ushort value) @property {
+ pool_.faultDisableCount = value;
+ }
+
+ /**
+ * The duration for which a client is no longer considered after it has
+ * failed too often.
+ *
+ * Defaults to one second.
+ */
+ Duration faultDisableDuration() const @property {
+ return pool_.faultDisableDuration;
+ }
+
+ /// Ditto
+ void faultDisableDuration(Duration value) @property {
+ pool_.faultDisableDuration = value;
+ }
+
+ /+override+/ bool delegate(Exception) rpcFaultFilter() const @property {
+ return rpcFaultFilter_;
+ }
+
+ /+override+/ void rpcFaultFilter(bool delegate(Exception) value) @property {
+ rpcFaultFilter_ = value;
+ }
+
+ /+override+/ bool reopenTransports() const @property {
+ return reopenTransports_;
+ }
+
+ /+override+/ void reopenTransports(bool value) @property {
+ reopenTransports_ = value;
+ }
+
+ mixin(fallbackPoolForwardCode!Interface());
+
+protected:
+ // The actual worker implementation to which RPC method calls are forwarded.
+ auto executeOnPool(string method, Args...)(Args args,
+ TCancellation cancellation
+ ) {
+ auto clients = pool_[];
+ if (clients.empty) {
+ throw new TException("No clients available to try.");
+ }
+
+ auto promise = new TPromise!(ReturnType!(MemberType!(Interface, method)));
+ Exception[] rpcExceptions;
+
+ void tryNext() {
+ while (clients.empty) {
+ Client next;
+ Duration waitTime;
+ if (clients.willBecomeNonempty(next, waitTime)) {
+ if (waitTime > dur!"hnsecs"(0)) {
+ if (waitTime < dur!"usecs"(10)) {
+ import core.thread;
+ Thread.sleep(waitTime);
+ } else {
+ next.transport.asyncManager.delay(waitTime, { tryNext(); });
+ return;
+ }
+ }
+ } else {
+ promise.fail(new TCompoundOperationException("All clients failed.",
+ rpcExceptions));
+ return;
+ }
+ }
+
+ auto client = clients.front;
+ clients.popFront;
+
+ if (reopenTransports) {
+ if (!client.transport.isOpen) {
+ try {
+ client.transport.open();
+ } catch (Exception e) {
+ pool_.recordFault(client);
+ tryNext();
+ return;
+ }
+ }
+ }
+
+ auto future = mixin("client." ~ method)(args, cancellation);
+ future.completion.addCallback({
+ if (future.status == TFutureStatus.CANCELLED) {
+ promise.cancel();
+ return;
+ }
+
+ auto e = future.getException();
+ if (e) {
+ if (rpcFaultFilter_ && rpcFaultFilter_(e)) {
+ pool_.recordFault(client);
+ rpcExceptions ~= e;
+ tryNext();
+ return;
+ }
+ }
+ pool_.recordSuccess(client);
+ promise.complete(future);
+ });
+ }
+
+ tryNext();
+ return promise;
+ }
+
+private:
+ TResourcePool!Client pool_;
+ bool delegate(Exception) rpcFaultFilter_;
+ bool reopenTransports_;
+}
+
+/**
+ * TAsyncClientPool construction helper to avoid having to explicitly
+ * specify the interface type, i.e. to allow the constructor being called
+ * using IFTI (see $(DMDBUG 6082, D Bugzilla enhancement request 6082)).
+ */
+TAsyncClientPool!Interface tAsyncClientPool(Interface)(
+ TAsyncClientBase!Interface[] clients
+) if (isService!Interface) {
+ return new typeof(return)(clients);
+}
+
+private {
+ // Cannot use an anonymous delegate literal for this because they aren't
+ // allowed in class scope.
+ string fallbackPoolForwardCode(Interface)() {
+ string code = "";
+
+ foreach (methodName; AllMemberMethodNames!Interface) {
+ enum qn = "Interface." ~ methodName;
+ code ~= "TFuture!(ReturnType!(" ~ qn ~ ")) " ~ methodName ~
+ "(ParameterTypeTuple!(" ~ qn ~ ") args, TCancellation cancellation = null) {\n";
+ code ~= "return executeOnPool!(`" ~ methodName ~ "`)(args, cancellation);\n";
+ code ~= "}\n";
+ }
+
+ return code;
+ }
+}
+
+/**
+ * A TAsyncClientPoolBase implementation which queries multiple servers at
+ * the same time and returns the first success response.
+ *
+ * The definition of »success« can be customized using the rpcFaultFilter()
+ * delegate property. If it is non-null and calling it for an exception set by
+ * a failed method invocation returns true, the error is considered to be
+ * caused by the RPC layer rather than the application layer, and the next
+ * server in the pool is tried. If all clients fail, the operation is marked
+ * as failed with a TCompoundOperationException.
+ */
+final class TAsyncFastestClientPool(Interface) if (isService!Interface) :
+ TAsyncClientPoolBase!Interface
+{
+ ///
+ this(Client[] clients) {
+ clients_ = clients;
+ rpcFaultFilter_ = defaultRpcFaultFilter;
+ reopenTransports_ = true;
+ }
+
+ /+override+/ void addClient(Client client) {
+ clients_ ~= client;
+ }
+
+ /+override+/ bool removeClient(Client client) {
+ auto oldLength = clients_.length;
+ clients_ = removeEqual(clients_, client);
+ return clients_.length < oldLength;
+ }
+
+
+ /+override+/ bool delegate(Exception) rpcFaultFilter() const @property {
+ return rpcFaultFilter_;
+ }
+
+ /+override+/ void rpcFaultFilter(bool delegate(Exception) value) @property {
+ rpcFaultFilter_ = value;
+ }
+
+ /+override+/bool reopenTransports() const @property {
+ return reopenTransports_;
+ }
+
+ /+override+/ void reopenTransports(bool value) @property {
+ reopenTransports_ = value;
+ }
+
+ mixin(fastestPoolForwardCode!Interface());
+
+private:
+ Client[] clients_;
+ bool delegate(Exception) rpcFaultFilter_;
+ bool reopenTransports_;
+}
+
+/**
+ * TAsyncFastestClientPool construction helper to avoid having to explicitly
+ * specify the interface type, i.e. to allow the constructor being called
+ * using IFTI (see $(DMDBUG 6082, D Bugzilla enhancement request 6082)).
+ */
+TAsyncFastestClientPool!Interface tAsyncFastestClientPool(Interface)(
+ TAsyncClientBase!Interface[] clients
+) if (isService!Interface) {
+ return new typeof(return)(clients);
+}
+
+private {
+ // Cannot use an anonymous delegate literal for this because they aren't
+ // allowed in class scope.
+ string fastestPoolForwardCode(Interface)() {
+ string code = "";
+
+ foreach (methodName; AllMemberMethodNames!Interface) {
+ enum qn = "Interface." ~ methodName;
+ code ~= "TFuture!(ReturnType!(" ~ qn ~ ")) " ~ methodName ~
+ "(ParameterTypeTuple!(" ~ qn ~ ") args, " ~
+ "TCancellation cancellation = null) {\n";
+ code ~= "enum methodName = `" ~ methodName ~ "`;\n";
+ code ~= q{
+ alias ReturnType!(MemberType!(Interface, methodName)) ResultType;
+
+ auto childCancellation = new TCancellationOrigin;
+
+ TFuture!ResultType[] futures;
+ futures.reserve(clients_.length);
+
+ foreach (c; clients_) {
+ if (reopenTransports) {
+ if (!c.transport.isOpen) {
+ try {
+ c.transport.open();
+ } catch (Exception e) {
+ continue;
+ }
+ }
+ }
+ futures ~= mixin("c." ~ methodName)(args, childCancellation);
+ }
+
+ return new FastestPoolJob!(ResultType)(
+ futures, rpcFaultFilter, cancellation, childCancellation);
+ };
+ code ~= "}\n";
+ }
+
+ return code;
+ }
+
+ final class FastestPoolJob(Result) : TFuture!Result {
+ this(TFuture!Result[] poolFutures, bool delegate(Exception) rpcFaultFilter,
+ TCancellation cancellation, TCancellationOrigin childCancellation
+ ) {
+ resultPromise_ = new TPromise!Result;
+ poolFutures_ = poolFutures;
+ rpcFaultFilter_ = rpcFaultFilter;
+ childCancellation_ = childCancellation;
+
+ foreach (future; poolFutures) {
+ future.completion.addCallback({
+ auto f = future;
+ return { completionCallback(f); };
+ }());
+ if (future.status != TFutureStatus.RUNNING) {
+ // If the current future is already completed, we are done, don't
+ // bother adding callbacks for the others (they would just return
+ // immediately after acquiring the lock).
+ return;
+ }
+ }
+
+ if (cancellation) {
+ cancellation.triggering.addCallback({
+ resultPromise_.cancel();
+ childCancellation.trigger();
+ });
+ }
+ }
+
+ TFutureStatus status() const @property {
+ return resultPromise_.status;
+ }
+
+ TAwaitable completion() @property {
+ return resultPromise_.completion;
+ }
+
+ Result get() {
+ return resultPromise_.get();
+ }
+
+ Exception getException() {
+ return resultPromise_.getException();
+ }
+
+ private:
+ void completionCallback(TFuture!Result future) {
+ synchronized {
+ if (future.status == TFutureStatus.CANCELLED) {
+ assert(resultPromise_.status != TFutureStatus.RUNNING);
+ return;
+ }
+
+ if (resultPromise_.status != TFutureStatus.RUNNING) {
+ // The operation has already been completed. This can happen if
+ // another client completed first, but this callback was already
+ // waiting for the lock when it called cancel().
+ return;
+ }
+
+ if (future.status == TFutureStatus.FAILED) {
+ auto e = future.getException();
+ if (rpcFaultFilter_ && rpcFaultFilter_(e)) {
+ rpcExceptions_ ~= e;
+
+ if (rpcExceptions_.length == poolFutures_.length) {
+ resultPromise_.fail(new TCompoundOperationException(
+ "All child operations failed, unable to retrieve a result.",
+ rpcExceptions_
+ ));
+ }
+
+ return;
+ }
+ }
+
+ // Store the result to the target promise.
+ resultPromise_.complete(future);
+
+ // Cancel the other futures, we would just discard their results.
+ // Note: We do this after we have stored the results to our promise,
+ // see the assert at the top of the function.
+ childCancellation_.trigger();
+ }
+ }
+
+ TPromise!Result resultPromise_;
+ TFuture!Result[] poolFutures_;
+ Exception[] rpcExceptions_;
+ bool delegate(Exception) rpcFaultFilter_;
+ TCancellationOrigin childCancellation_;
+ }
+}
+
+/**
+ * Allows easily aggregating results from a number of TAsyncClients.
+ *
+ * Contrary to TAsync{Fallback, Fastest}ClientPool, this class does not
+ * simply implement TFutureInterface!Interface. It manages a pool of clients,
+ * but allows the user to specify a custom accumulator function to use or to
+ * iterate over the results using a TFutureAggregatorRange.
+ *
+ * For each service method, TAsyncAggregator offers a method
+ * accepting the same arguments, and an optional TCancellation instance, just
+ * like with TFutureInterface. The return type, however, is a proxy object
+ * that offers the following methods:
+ * ---
+ * /++
+ * + Returns a thrift.util.future.TFutureAggregatorRange for the results of
+ * + the client pool method invocations.
+ * +
+ * + The [] (slicing) operator can also be used to obtain the range.
+ * +
+ * + Params:
+ * + timeout = A timeout to pass to the TFutureAggregatorRange constructor,
+ * + defaults to zero (no timeout).
+ * +/
+ * TFutureAggregatorRange!ReturnType range(Duration timeout = dur!"hnsecs"(0));
+ * auto opSlice() { return range(); } /// Ditto
+ *
+ * /++
+ * + Returns a future that gathers the results from the clients in the pool
+ * + and invokes a user-supplied accumulator function on them, returning its
+ * + return value to the client.
+ * +
+ * + In addition to the TFuture!AccumulatedType interface (where
+ * + AccumulatedType is the return type of the accumulator function), the
+ * + returned object also offers two additional methods, finish() and
+ * + finishGet(): By default, the accumulator functions is called after all
+ * + the results from the pool clients have become available. Calling finish()
+ * + causes the accumulator future to stop waiting for other results and
+ * + immediately invoking the accumulator function on the results currently
+ * + available. If all results are already available, finish() is a no-op.
+ * + finishGet() is a convenience shortcut for combining it with
+ * + a call to get() immediately afterwards, like waitGet() is for wait().
+ * +
+ * + The acc alias can point to any callable accepting either an array of
+ * + return values or an array of return values and an array of exceptions;
+ * + see isAccumulator!() for details. The default accumulator concatenates
+ * + return values that can be concatenated with each others (e.g. arrays),
+ * + and simply returns an array of values otherwise, failing with a
+ * + TCompoundOperationException no values were returned.
+ * +
+ * + The accumulator function is not executed in any of the async manager
+ * + worker threads associated with the async clients, but instead it is
+ * + invoked when the actual result is requested for the first time after the
+ * + operation has been completed. This also includes checking the status
+ * + of the operation once it is no longer running, since the accumulator
+ * + has to be run to determine whether the operation succeeded or failed.
+ * +/
+ * auto accumulate(alias acc = defaultAccumulator)() if (isAccumulator!acc);
+ * ---
+ *
+ * Example:
+ * ---
+ * // Some Thrift service.
+ * interface Foo {
+ * int foo(string name);
+ * byte[] bar();
+ * }
+ *
+ * // Create the aggregator pool – client0, client1, client2 are some
+ * // TAsyncClient!Foo instances, but in theory could also be other
+ * // TFutureInterface!Foo implementations (e.g. some async client pool).
+ * auto pool = new TAsyncAggregator!Foo([client0, client1, client2]);
+ *
+ * foreach (val; pool.foo("baz").range(dur!"seconds"(1))) {
+ * // Process all the results that are available before a second has passed,
+ * // in the order they arrive.
+ * writeln(val);
+ * }
+ *
+ * auto sumRoots = pool.foo("baz").accumulate!((int[] vals, Exceptions[] exs){
+ * if (vals.empty) {
+ * throw new TCompoundOperationException("All clients failed", exs);
+ * }
+ *
+ * // Just to illustrate that the type of the values can change, convert the
+ * // numbers to double and sum up their roots.
+ * double result = 0;
+ * foreach (v; vals) result += sqrt(cast(double)v);
+ * return result;
+ * })();
+ *
+ * // Wait up to three seconds for the result, and then accumulate what has
+ * // arrived so far.
+ * sumRoots.completion.wait(dur!"seconds"(3));
+ * writeln(sumRoots.finishGet());
+ *
+ * // For scalars, the default accumulator returns an array of the values.
+ * pragma(msg, typeof(pool.foo("").accumulate().get()); // int[].
+ *
+ * // For lists, etc., it concatenates the results together.
+ * pragma(msg, typeof(pool.bar().accumulate().get())); // byte[].
+ * ---
+ *
+ * Note: For the accumulate!() interface, you might currently hit a »cannot use
+ * local '…' as parameter to non-global template accumulate«-error, see
+ * $(DMDBUG 5710, DMD issue 5710). If your accumulator function does not need
+ * to access the surrounding scope, you might want to use a function literal
+ * instead of a delegate to avoid the issue.
+ */
+class TAsyncAggregator(Interface) if (isBaseService!Interface) {
+ /// Shorthand for the client type this instance operates on.
+ alias TAsyncClientBase!Interface Client;
+
+ ///
+ this(Client[] clients) {
+ clients_ = clients;
+ }
+
+ /// Whether to open the underlying transports of a client before trying to
+ /// execute a method if they are not open. This is usually desirable
+ /// because it allows e.g. to automatically reconnect to a remote server
+ /// if the network connection is dropped.
+ ///
+ /// Defaults to true.
+ bool reopenTransports = true;
+
+ mixin AggregatorOpDispatch!();
+
+private:
+ Client[] clients_;
+}
+
+/// Ditto
+class TAsyncAggregator(Interface) if (isDerivedService!Interface) :
+ TAsyncAggregator!(BaseService!Interface)
+{
+ /// Shorthand for the client type this instance operates on.
+ alias TAsyncClientBase!Interface Client;
+
+ ///
+ this(Client[] clients) {
+ super(cast(TAsyncClientBase!(BaseService!Interface)[])clients);
+ }
+
+ mixin AggregatorOpDispatch!();
+}
+
+/**
+ * Whether fun is a valid accumulator function for values of type ValueType.
+ *
+ * For this to be true, fun must be a callable matching one of the following
+ * argument lists:
+ * ---
+ * fun(ValueType[] values);
+ * fun(ValueType[] values, Exception[] exceptions);
+ * ---
+ *
+ * The second version is passed the collected array exceptions from all the
+ * clients in the pool.
+ *
+ * The return value of the accumulator function is passed to the client (via
+ * the result future). If it throws an exception, the operation is marked as
+ * failed with the given exception instead.
+ */
+template isAccumulator(ValueType, alias fun) {
+ enum isAccumulator = is(typeof(fun(cast(ValueType[])[]))) ||
+ is(typeof(fun(cast(ValueType[])[], cast(Exception[])[])));
+}
+
+/**
+ * TAsyncAggregator construction helper to avoid having to explicitly
+ * specify the interface type, i.e. to allow the constructor being called
+ * using IFTI (see $(DMDBUG 6082, D Bugzilla enhancement request 6082)).
+ */
+TAsyncAggregator!Interface tAsyncAggregator(Interface)(
+ TAsyncClientBase!Interface[] clients
+) if (isService!Interface) {
+ return new typeof(return)(clients);
+}
+
+private {
+ mixin template AggregatorOpDispatch() {
+ auto opDispatch(string name, Args...)(Args args) if (
+ is(typeof(mixin("Interface.init." ~ name)(args)))
+ ) {
+ alias ReturnType!(MemberType!(Interface, name)) ResultType;
+
+ auto childCancellation = new TCancellationOrigin;
+
+ TFuture!ResultType[] futures;
+ futures.reserve(clients_.length);
+
+ foreach (c; cast(Client[])clients_) {
+ if (reopenTransports) {
+ if (!c.transport.isOpen) {
+ try {
+ c.transport.open();
+ } catch (Exception e) {
+ continue;
+ }
+ }
+ }
+ futures ~= mixin("c." ~ name)(args, childCancellation);
+ }
+
+ return AggregationResult!ResultType(futures, childCancellation);
+ }
+ }
+
+ struct AggregationResult(T) {
+ auto opSlice() {
+ return range();
+ }
+
+ auto range(Duration timeout = dur!"hnsecs"(0)) {
+ return tFutureAggregatorRange(futures_, childCancellation_, timeout);
+ }
+
+ auto accumulate(alias acc = defaultAccumulator)() if (isAccumulator!(T, acc)) {
+ return new AccumulatorJob!(T, acc)(futures_, childCancellation_);
+ }
+
+ private:
+ TFuture!T[] futures_;
+ TCancellationOrigin childCancellation_;
+ }
+
+ auto defaultAccumulator(T)(T[] values, Exception[] exceptions) {
+ if (values.empty) {
+ throw new TCompoundOperationException("All clients failed",
+ exceptions);
+ }
+
+ static if (is(typeof(T.init ~ T.init))) {
+ import std.algorithm;
+ return reduce!"a ~ b"(values);
+ } else {
+ return values;
+ }
+ }
+
+ final class AccumulatorJob(T, alias accumulator) if (
+ isAccumulator!(T, accumulator)
+ ) : TFuture!(AccumulatorResult!(T, accumulator)) {
+ this(TFuture!T[] futures, TCancellationOrigin childCancellation) {
+ futures_ = futures;
+ childCancellation_ = childCancellation;
+ resultMutex_ = new Mutex;
+ completionEvent_ = new TOneshotEvent;
+
+ foreach (future; futures) {
+ future.completion.addCallback({
+ auto f = future;
+ return {
+ synchronized (resultMutex_) {
+ if (f.status == TFutureStatus.CANCELLED) {
+ if (!finished_) {
+ status_ = TFutureStatus.CANCELLED;
+ finished_ = true;
+ }
+ return;
+ }
+
+ if (f.status == TFutureStatus.FAILED) {
+ exceptions_ ~= f.getException();
+ } else {
+ results_ ~= f.get();
+ }
+
+ if (results_.length + exceptions_.length == futures_.length) {
+ finished_ = true;
+ completionEvent_.trigger();
+ }
+ }
+ };
+ }());
+ }
+ }
+
+ TFutureStatus status() @property {
+ synchronized (resultMutex_) {
+ if (!finished_) return TFutureStatus.RUNNING;
+ if (status_ != TFutureStatus.RUNNING) return status_;
+
+ try {
+ result_ = invokeAccumulator!accumulator(results_, exceptions_);
+ status_ = TFutureStatus.SUCCEEDED;
+ } catch (Exception e) {
+ exception_ = e;
+ status_ = TFutureStatus.FAILED;
+ }
+
+ return status_;
+ }
+ }
+
+ TAwaitable completion() @property {
+ return completionEvent_;
+ }
+
+ AccumulatorResult!(T, accumulator) get() {
+ auto s = status;
+
+ enforce(s != TFutureStatus.RUNNING,
+ new TFutureException("Operation not yet completed."));
+
+ if (s == TFutureStatus.CANCELLED) throw new TCancelledException;
+ if (s == TFutureStatus.FAILED) throw exception_;
+ return result_;
+ }
+
+ Exception getException() {
+ auto s = status;
+ enforce(s != TFutureStatus.RUNNING,
+ new TFutureException("Operation not yet completed."));
+
+ if (s == TFutureStatus.CANCELLED) throw new TCancelledException;
+
+ if (s == TFutureStatus.SUCCEEDED) {
+ return null;
+ }
+ return exception_;
+ }
+
+ void finish() {
+ synchronized (resultMutex_) {
+ if (!finished_) {
+ finished_ = true;
+ childCancellation_.trigger();
+ completionEvent_.trigger();
+ }
+ }
+ }
+
+ auto finishGet() {
+ finish();
+ return get();
+ }
+
+ private:
+ TFuture!T[] futures_;
+ TCancellationOrigin childCancellation_;
+
+ bool finished_;
+ T[] results_;
+ Exception[] exceptions_;
+
+ TFutureStatus status_;
+ Mutex resultMutex_;
+ union {
+ AccumulatorResult!(T, accumulator) result_;
+ Exception exception_;
+ }
+ TOneshotEvent completionEvent_;
+ }
+
+ auto invokeAccumulator(alias accumulator, T)(
+ T[] values, Exception[] exceptions
+ ) if (
+ isAccumulator!(T, accumulator)
+ ) {
+ static if (is(typeof(accumulator(values, exceptions)))) {
+ return accumulator(values, exceptions);
+ } else {
+ return accumulator(values);
+ }
+ }
+
+ template AccumulatorResult(T, alias acc) {
+ alias typeof(invokeAccumulator!acc(cast(T[])[], cast(Exception[])[]))
+ AccumulatorResult;
+ }
+}
projects / ceph.git / blobdiff