--- /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.
+ */
+
+/**
+ * Defines the interface used for client-side handling of asynchronous
+ * I/O operations, based on coroutines.
+ *
+ * The main piece of the »client side« (e.g. for TAsyncClient users) of the
+ * API is TFuture, which represents an asynchronously executed operation,
+ * which can have a return value, throw exceptions, and which can be waited
+ * upon.
+ *
+ * On the »implementation side«, the idea is that by using a TAsyncTransport
+ * instead of a normal TTransport and executing the work through a
+ * TAsyncManager, the same code as for synchronous I/O can be used for
+ * asynchronous operation as well, for example:
+ *
+ * ---
+ * auto socket = new TAsyncSocket(someTAsyncSocketManager(), host, port);
+ * // …
+ * socket.asyncManager.execute(socket, {
+ * SomeThriftStruct s;
+ *
+ * // Waiting for socket I/O will not block an entire thread but cause
+ * // the async manager to execute another task in the meantime, because
+ * // we are using TAsyncSocket instead of TSocket.
+ * s.read(socket);
+ *
+ * // Do something with s, e.g. set a TPromise result to it.
+ * writeln(s);
+ * });
+ * ---
+ */
+module thrift.async.base;
+
+import core.time : Duration, dur;
+import std.socket/+ : Socket+/; // DMD @@BUG314@@
+import thrift.base;
+import thrift.transport.base;
+import thrift.util.cancellation;
+
+/**
+ * Manages one or more asynchronous transport resources (e.g. sockets in the
+ * case of TAsyncSocketManager) and allows work items to be submitted for them.
+ *
+ * Implementations will typically run one or more background threads for
+ * executing the work, which is one of the reasons for a TAsyncManager to be
+ * used. Each work item is run in its own fiber and is expected to yield() away
+ * while waiting for time-consuming operations.
+ *
+ * The second important purpose of TAsyncManager is to serialize access to
+ * the transport resources – without taking care of that, e.g. issuing multiple
+ * RPC calls over the same connection in rapid succession would likely lead to
+ * more than one request being written at the same time, causing only garbage
+ * to arrive at the remote end.
+ *
+ * All methods are thread-safe.
+ */
+interface TAsyncManager {
+ /**
+ * Submits a work item to be executed asynchronously.
+ *
+ * Access to asnyc transports is serialized – if two work items associated
+ * with the same transport are submitted, the second delegate will not be
+ * invoked until the first has returned, even it the latter context-switches
+ * away (because it is waiting for I/O) and the async manager is idle
+ * otherwise.
+ *
+ * Optionally, a TCancellation instance can be specified. If present,
+ * triggering it will be considered a request to cancel the work item, if it
+ * is still waiting for the associated transport to become available.
+ * Delegates which are already being processed (i.e. waiting for I/O) are not
+ * affected because this would bring the connection into an undefined state
+ * (as probably half-written request or a half-read response would be left
+ * behind).
+ *
+ * Params:
+ * transport = The TAsyncTransport the work delegate will operate on. Must
+ * be associated with this TAsyncManager instance.
+ * work = The operations to execute on the given transport. Must never
+ * throw, errors should be handled in another way. nothrow semantics are
+ * difficult to enforce in combination with fibres though, so currently
+ * exceptions are just swallowed by TAsyncManager implementations.
+ * cancellation = If set, can be used to request cancellatinon of this work
+ * item if it is still waiting to be executed.
+ *
+ * Note: The work item will likely be executed in a different thread, so make
+ * sure the code it relies on is thread-safe. An exception are the async
+ * transports themselves, to which access is serialized as noted above.
+ */
+ void execute(TAsyncTransport transport, void delegate() work,
+ TCancellation cancellation = null
+ ) in {
+ assert(transport.asyncManager is this,
+ "The given transport must be associated with this TAsyncManager.");
+ }
+
+ /**
+ * Submits a delegate to be executed after a certain amount of time has
+ * passed.
+ *
+ * The actual amount of time elapsed can be higher if the async manager
+ * instance is busy and thus should not be relied on. The
+ *
+ * Params:
+ * duration = The amount of time to wait before starting to execute the
+ * work delegate.
+ * work = The code to execute after the specified amount of time has passed.
+ *
+ * Example:
+ * ---
+ * // A very basic example – usually, the actuall work item would enqueue
+ * // some async transport operation.
+ * auto asyncMangager = someAsyncManager();
+ *
+ * TFuture!int calculate() {
+ * // Create a promise and asynchronously set its value after three
+ * // seconds have passed.
+ * auto promise = new TPromise!int;
+ * asyncManager.delay(dur!"seconds"(3), {
+ * promise.succeed(42);
+ * });
+ *
+ * // Immediately return it to the caller.
+ * return promise;
+ * }
+ *
+ * // This will wait until the result is available and then print it.
+ * writeln(calculate().waitGet());
+ * ---
+ */
+ void delay(Duration duration, void delegate() work);
+
+ /**
+ * Shuts down all background threads or other facilities that might have
+ * been started in order to execute work items. This function is typically
+ * called during program shutdown.
+ *
+ * If there are still tasks to be executed when the timeout expires, any
+ * currently executed work items will never receive any notifications
+ * for async transports managed by this instance, queued work items will
+ * be silently dropped, and implementations are allowed to leak resources.
+ *
+ * Params:
+ * waitFinishTimeout = If positive, waits for all work items to be
+ * finished for the specified amount of time, if negative, waits for
+ * completion without ever timing out, if zero, immediately shuts down
+ * the background facilities.
+ */
+ bool stop(Duration waitFinishTimeout = dur!"hnsecs"(-1));
+}
+
+/**
+ * A TTransport which uses a TAsyncManager to schedule non-blocking operations.
+ *
+ * The actual type of device is not specified; typically, implementations will
+ * depend on an interface derived from TAsyncManager to be notified of changes
+ * in the transport state.
+ *
+ * The peeking, reading, writing and flushing methods must always be called
+ * from within the associated async manager.
+ */
+interface TAsyncTransport : TTransport {
+ /**
+ * The TAsyncManager associated with this transport.
+ */
+ TAsyncManager asyncManager() @property;
+}
+
+/**
+ * A TAsyncManager providing notificiations for socket events.
+ */
+interface TAsyncSocketManager : TAsyncManager {
+ /**
+ * Adds a listener that is triggered once when an event of the specified type
+ * occurs, and removed afterwards.
+ *
+ * Params:
+ * socket = The socket to listen for events at.
+ * eventType = The type of the event to listen for.
+ * timeout = The period of time after which the listener will be called
+ * with TAsyncEventReason.TIMED_OUT if no event happened.
+ * listener = The delegate to call when an event happened.
+ */
+ void addOneshotListener(Socket socket, TAsyncEventType eventType,
+ Duration timeout, TSocketEventListener listener);
+
+ /// Ditto
+ void addOneshotListener(Socket socket, TAsyncEventType eventType,
+ TSocketEventListener listener);
+}
+
+/**
+ * Types of events that can happen for an asynchronous transport.
+ */
+enum TAsyncEventType {
+ READ, /// New data became available to read.
+ WRITE /// The transport became ready to be written to.
+}
+
+/**
+ * The type of the delegates used to register socket event handlers.
+ */
+alias void delegate(TAsyncEventReason callReason) TSocketEventListener;
+
+/**
+ * The reason a listener was called.
+ */
+enum TAsyncEventReason : byte {
+ NORMAL, /// The event listened for was triggered normally.
+ TIMED_OUT /// A timeout for the event was set, and it expired.
+}
projects / ceph.git / blobdiff