lib/d/src/thrift/async/socket.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.async.socket;

 import core.thread : Fiber;
 import core.time : dur, Duration;
 import std.array : empty;
 import std.conv : to;
 import std.exception : enforce;
 import std.socket;
 import thrift.base;
 import thrift.async.base;
 import thrift.transport.base;
 import thrift.transport.socket : TSocketBase;
 import thrift.internal.endian;
 import thrift.internal.socket;

 version (Windows) {
   import std.c.windows.winsock : connect;
 } else version (Posix) {
   import core.sys.posix.sys.socket : connect;
 } else static assert(0, "Don't know connect on this platform.");

 /**
  * Non-blocking socket implementation of the TTransport interface.
  *
  * Whenever a socket operation would block, TAsyncSocket registers a callback
  * with the specified TAsyncSocketManager and yields.
  *
  * As for thrift.transport.socket, due to the limitations of std.socket,
  * currently only TCP/IP sockets are supported (i.e. Unix domain sockets are
  * not).
  */
 class TAsyncSocket : TSocketBase, TAsyncTransport {
   /**
    * Constructor that takes an already created, connected (!) socket.
    *
    * Params:
    *   asyncManager = The TAsyncSocketManager to use for non-blocking I/O.
    *   socket = Already created, connected socket object. Will be switched to
    *     non-blocking mode if it isn't already.
    */
   this(TAsyncSocketManager asyncManager, Socket socket) {
     asyncManager_ = asyncManager;
     socket.blocking = false;
     super(socket);
   }

   /**
    * Creates a new unconnected socket that will connect to the given host
    * on the given port.
    *
    * Params:
    *   asyncManager = The TAsyncSocketManager to use for non-blocking I/O.
    *   host = Remote host.
    *   port = Remote port.
    */
   this(TAsyncSocketManager asyncManager, string host, ushort port) {
     asyncManager_ = asyncManager;
     super(host, port);
   }

   override TAsyncManager asyncManager() @property {
     return asyncManager_;
   }

   /**
    * Asynchronously connects the socket.
    *
    * Completes without blocking and defers further operations on the socket
    * until the connection is established. If connecting fails, this is
    * currently not indicated in any way other than every call to read/write
    * failing.
    */
   override void open() {
     if (isOpen) return;

     enforce(!host_.empty, new TTransportException(
       "Cannot open null host.", TTransportException.Type.NOT_OPEN));
     enforce(port_ != 0, new TTransportException(
       "Cannot open with null port.", TTransportException.Type.NOT_OPEN));


     // Cannot use std.socket.Socket.connect here because it hides away
     // EINPROGRESS/WSAWOULDBLOCK.
     Address addr;
     try {
       // Currently, we just go with the first address returned, could be made
       // more intelligent though – IPv6?
       addr = getAddress(host_, port_)[0];
     } catch (Exception e) {
       throw new TTransportException(`Unable to resolve host "` ~ host_ ~ `".`,
         TTransportException.Type.NOT_OPEN, __FILE__, __LINE__, e);
     }

     socket_ = new TcpSocket(addr.addressFamily);
     socket_.blocking = false;
     setSocketOpts();

     auto errorCode = connect(socket_.handle, addr.name(), addr.nameLen());
     if (errorCode == 0) {
       // If the connection could be established immediately, just return. I
       // don't know if this ever happens.
       return;
     }

     auto errno = getSocketErrno();
     if (errno != CONNECT_INPROGRESS_ERRNO) {
       throw new TTransportException(`Could not establish connection to "` ~
         host_ ~ `": ` ~ socketErrnoString(errno),
         TTransportException.Type.NOT_OPEN);
     }

     // This is the expected case: connect() signalled that the connection
     // is being established in the background. Queue up a work item with the
     // async manager which just defers any other operations on this
     // TAsyncSocket instance until the socket is ready.
     asyncManager_.execute(this,
       {
         auto fiber = Fiber.getThis();
         TAsyncEventReason reason = void;
         asyncManager_.addOneshotListener(socket_, TAsyncEventType.WRITE,
           connectTimeout,
           scopedDelegate((TAsyncEventReason r){ reason = r; fiber.call(); })
         );
         Fiber.yield();

         if (reason == TAsyncEventReason.TIMED_OUT) {
           // Close the connection, so that subsequent work items fail immediately.
           closeImmediately();
           return;
         }

         int errorCode = void;
         socket_.getOption(SocketOptionLevel.SOCKET, cast(SocketOption)SO_ERROR,
           errorCode);

         if (errorCode) {
           logInfo("Could not connect TAsyncSocket: %s",
             socketErrnoString(errorCode));

           // Close the connection, so that subsequent work items fail immediately.
           closeImmediately();
           return;
         }

       }
     );
   }

   /**
    * Closes the socket.
    *
    * Will block until all currently active operations are finished before the
    * socket is closed.
    */
   override void close() {
     if (!isOpen) return;

     import core.sync.condition;
     import core.sync.mutex;

     auto doneMutex = new Mutex;
     auto doneCond = new Condition(doneMutex);
     synchronized (doneMutex) {
       asyncManager_.execute(this,
         scopedDelegate(
           {
             closeImmediately();
             synchronized (doneMutex) doneCond.notifyAll();
           }
         )
       );
       doneCond.wait();
     }
   }

   override bool peek() {
     if (!isOpen) return false;

     ubyte buf;
     auto r = socket_.receive((&buf)[0..1], SocketFlags.PEEK);
     if (r == Socket.ERROR) {
       auto lastErrno = getSocketErrno();
       static if (connresetOnPeerShutdown) {
         if (lastErrno == ECONNRESET) {
           closeImmediately();
           return false;
         }
       }
       throw new TTransportException("Peeking into socket failed: " ~
         socketErrnoString(lastErrno), TTransportException.Type.UNKNOWN);
     }
     return (r > 0);
   }

   override size_t read(ubyte[] buf) {
     enforce(isOpen, new TTransportException(
       "Cannot read if socket is not open.", TTransportException.Type.NOT_OPEN));

     typeof(getSocketErrno()) lastErrno;

     auto r = yieldOnBlock(socket_.receive(cast(void[])buf),
       TAsyncEventType.READ);

     // If recv went fine, immediately return.
     if (r >= 0) return r;

     // Something went wrong, find out how to handle it.
     lastErrno = getSocketErrno();

     static if (connresetOnPeerShutdown) {
       // See top comment.
       if (lastErrno == ECONNRESET) {
         return 0;
       }
     }

     throw new TTransportException("Receiving from socket failed: " ~
       socketErrnoString(lastErrno), TTransportException.Type.UNKNOWN);
   }

   override void write(in ubyte[] buf) {
     size_t sent;
     while (sent < buf.length) {
       sent += writeSome(buf[sent .. $]);
     }
     assert(sent == buf.length);
   }

   override size_t writeSome(in ubyte[] buf) {
     enforce(isOpen, new TTransportException(
       "Cannot write if socket is not open.", TTransportException.Type.NOT_OPEN));

     auto r = yieldOnBlock(socket_.send(buf), TAsyncEventType.WRITE);

     // Everything went well, just return the number of bytes written.
     if (r > 0) return r;

     // Handle error conditions.
     if (r < 0) {
       auto lastErrno = getSocketErrno();

       auto type = TTransportException.Type.UNKNOWN;
       if (isSocketCloseErrno(lastErrno)) {
         type = TTransportException.Type.NOT_OPEN;
         closeImmediately();
       }

       throw new TTransportException("Sending to socket failed: " ~
         socketErrnoString(lastErrno), type);
     }

     // send() should never return 0.
     throw new TTransportException("Sending to socket failed (0 bytes written).",
       TTransportException.Type.UNKNOWN);
   }

   /// The amount of time in which a conncetion must be established before the
   /// open() call times out.
   Duration connectTimeout = dur!"seconds"(5);

 private:
   void closeImmediately() {
     socket_.close();
     socket_ = null;
   }

   T yieldOnBlock(T)(lazy T call, TAsyncEventType eventType) {
     while (true) {
       auto result = call();
       if (result != Socket.ERROR || getSocketErrno() != WOULD_BLOCK_ERRNO) return result;

       // We got an EAGAIN result, register a callback to return here once some
       // event happens and yield.

       Duration timeout = void;
       final switch (eventType) {
         case TAsyncEventType.READ:
           timeout = recvTimeout_;
           break;
         case TAsyncEventType.WRITE:
           timeout = sendTimeout_;
           break;
       }

       auto fiber = Fiber.getThis();
       assert(fiber, "Current fiber null – not running in TAsyncManager?");
       TAsyncEventReason eventReason = void;
       asyncManager_.addOneshotListener(socket_, eventType, timeout,
         scopedDelegate((TAsyncEventReason reason) {
           eventReason = reason;
           fiber.call();
         })
       );

       // Yields execution back to the async manager, will return back here once
       // the above listener is called.
       Fiber.yield();

       if (eventReason == TAsyncEventReason.TIMED_OUT) {
         // If we are cancelling the request due to a timed out operation, the
         // connection is in an undefined state, because the server could decide
         // to send the requested data later, or we could have already been half-
         // way into writing a request. Thus, we close the connection to make any
         // possibly queued up work items fail immediately. Besides, the server
         // is not very likely to immediately recover after a socket-level
         // timeout has expired anyway.
         closeImmediately();

         throw new TTransportException("Timed out while waiting for socket " ~
           "to get ready to " ~ to!string(eventType) ~ ".",
           TTransportException.Type.TIMED_OUT);
       }
     }
   }

   /// The TAsyncSocketManager to use for non-blocking I/O.
   TAsyncSocketManager asyncManager_;
 }

 private {
   // std.socket doesn't include SO_ERROR for reasons unknown.
   version (linux) {
     enum SO_ERROR = 4;
   } else version (OSX) {
     enum SO_ERROR = 0x1007;
   } else version (FreeBSD) {
     enum SO_ERROR = 0x1007;
   } else version (Win32) {
     import std.c.windows.winsock : SO_ERROR;
   } else static assert(false, "Don't know SO_ERROR on this platform.");

   // This hack forces a delegate literal to be scoped, even if it is passed to
   // a function accepting normal delegates as well. DMD likes to allocate the
   // context on the heap anyway, but it seems to work for LDC.
   import std.traits : isDelegate;
   auto scopedDelegate(D)(scope D d) if (isDelegate!D) {
     return d;
   }
 }
	/*
	* 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.async.socket;

	import core.thread : Fiber;
	import core.time : dur, Duration;
	import std.array : empty;
	import std.conv : to;
	import std.exception : enforce;
	import std.socket;
	import thrift.base;
	import thrift.async.base;
	import thrift.transport.base;
	import thrift.transport.socket : TSocketBase;
	import thrift.internal.endian;
	import thrift.internal.socket;

	version (Windows) {
	import std.c.windows.winsock : connect;
	} else version (Posix) {
	import core.sys.posix.sys.socket : connect;
	} else static assert(0, "Don't know connect on this platform.");

	/**
	* Non-blocking socket implementation of the TTransport interface.
	*
	* Whenever a socket operation would block, TAsyncSocket registers a callback
	* with the specified TAsyncSocketManager and yields.
	*
	* As for thrift.transport.socket, due to the limitations of std.socket,
	* currently only TCP/IP sockets are supported (i.e. Unix domain sockets are
	* not).
	*/
	class TAsyncSocket : TSocketBase, TAsyncTransport {
	/**
	* Constructor that takes an already created, connected (!) socket.
	*
	* Params:
	* asyncManager = The TAsyncSocketManager to use for non-blocking I/O.
	* socket = Already created, connected socket object. Will be switched to
	* non-blocking mode if it isn't already.
	*/
	this(TAsyncSocketManager asyncManager, Socket socket) {
	asyncManager_ = asyncManager;
	socket.blocking = false;
	super(socket);
	}

	/**
	* Creates a new unconnected socket that will connect to the given host
	* on the given port.
	*
	* Params:
	* asyncManager = The TAsyncSocketManager to use for non-blocking I/O.
	* host = Remote host.
	* port = Remote port.
	*/
	this(TAsyncSocketManager asyncManager, string host, ushort port) {
	asyncManager_ = asyncManager;
	super(host, port);
	}

	override TAsyncManager asyncManager() @property {
	return asyncManager_;
	}

	/**
	* Asynchronously connects the socket.
	*
	* Completes without blocking and defers further operations on the socket
	* until the connection is established. If connecting fails, this is
	* currently not indicated in any way other than every call to read/write
	* failing.
	*/
	override void open() {
	if (isOpen) return;

	enforce(!host_.empty, new TTransportException(
	"Cannot open null host.", TTransportException.Type.NOT_OPEN));
	enforce(port_ != 0, new TTransportException(
	"Cannot open with null port.", TTransportException.Type.NOT_OPEN));


	// Cannot use std.socket.Socket.connect here because it hides away
	// EINPROGRESS/WSAWOULDBLOCK.
	Address addr;
	try {
	// Currently, we just go with the first address returned, could be made
	// more intelligent though – IPv6?
	addr = getAddress(host_, port_)[0];
	} catch (Exception e) {
	throw new TTransportException(`Unable to resolve host "` ~ host_ ~ `".`,
	TTransportException.Type.NOT_OPEN, __FILE__, __LINE__, e);
	}

	socket_ = new TcpSocket(addr.addressFamily);
	socket_.blocking = false;
	setSocketOpts();

	auto errorCode = connect(socket_.handle, addr.name(), addr.nameLen());
	if (errorCode == 0) {
	// If the connection could be established immediately, just return. I
	// don't know if this ever happens.
	return;
	}

	auto errno = getSocketErrno();
	if (errno != CONNECT_INPROGRESS_ERRNO) {
	throw new TTransportException(`Could not establish connection to "` ~
	host_ ~ `": ` ~ socketErrnoString(errno),
	TTransportException.Type.NOT_OPEN);
	}

	// This is the expected case: connect() signalled that the connection
	// is being established in the background. Queue up a work item with the
	// async manager which just defers any other operations on this
	// TAsyncSocket instance until the socket is ready.
	asyncManager_.execute(this,
	{
	auto fiber = Fiber.getThis();
	TAsyncEventReason reason = void;
	asyncManager_.addOneshotListener(socket_, TAsyncEventType.WRITE,
	connectTimeout,
	scopedDelegate((TAsyncEventReason r){ reason = r; fiber.call(); })
	);
	Fiber.yield();

	if (reason == TAsyncEventReason.TIMED_OUT) {
	// Close the connection, so that subsequent work items fail immediately.
	closeImmediately();
	return;
	}

	int errorCode = void;
	socket_.getOption(SocketOptionLevel.SOCKET, cast(SocketOption)SO_ERROR,
	errorCode);

	if (errorCode) {
	logInfo("Could not connect TAsyncSocket: %s",
	socketErrnoString(errorCode));

	// Close the connection, so that subsequent work items fail immediately.
	closeImmediately();
	return;
	}

	}
	);
	}

	/**
	* Closes the socket.
	*
	* Will block until all currently active operations are finished before the
	* socket is closed.
	*/
	override void close() {
	if (!isOpen) return;

	import core.sync.condition;
	import core.sync.mutex;

	auto doneMutex = new Mutex;
	auto doneCond = new Condition(doneMutex);
	synchronized (doneMutex) {
	asyncManager_.execute(this,
	scopedDelegate(
	{
	closeImmediately();
	synchronized (doneMutex) doneCond.notifyAll();
	}
	)
	);
	doneCond.wait();
	}
	}

	override bool peek() {
	if (!isOpen) return false;

	ubyte buf;
	auto r = socket_.receive((&buf)[0..1], SocketFlags.PEEK);
	if (r == Socket.ERROR) {
	auto lastErrno = getSocketErrno();
	static if (connresetOnPeerShutdown) {
	if (lastErrno == ECONNRESET) {
	closeImmediately();
	return false;
	}
	}
	throw new TTransportException("Peeking into socket failed: " ~
	socketErrnoString(lastErrno), TTransportException.Type.UNKNOWN);
	}
	return (r > 0);
	}

	override size_t read(ubyte[] buf) {
	enforce(isOpen, new TTransportException(
	"Cannot read if socket is not open.", TTransportException.Type.NOT_OPEN));

	typeof(getSocketErrno()) lastErrno;

	auto r = yieldOnBlock(socket_.receive(cast(void[])buf),
	TAsyncEventType.READ);

	// If recv went fine, immediately return.
	if (r >= 0) return r;

	// Something went wrong, find out how to handle it.
	lastErrno = getSocketErrno();

	static if (connresetOnPeerShutdown) {
	// See top comment.
	if (lastErrno == ECONNRESET) {
	return 0;
	}
	}

	throw new TTransportException("Receiving from socket failed: " ~
	socketErrnoString(lastErrno), TTransportException.Type.UNKNOWN);
	}

	override void write(in ubyte[] buf) {
	size_t sent;
	while (sent < buf.length) {
	sent += writeSome(buf[sent .. $]);
	}
	assert(sent == buf.length);
	}

	override size_t writeSome(in ubyte[] buf) {
	enforce(isOpen, new TTransportException(
	"Cannot write if socket is not open.", TTransportException.Type.NOT_OPEN));

	auto r = yieldOnBlock(socket_.send(buf), TAsyncEventType.WRITE);

	// Everything went well, just return the number of bytes written.
	if (r > 0) return r;

	// Handle error conditions.
	if (r < 0) {
	auto lastErrno = getSocketErrno();

	auto type = TTransportException.Type.UNKNOWN;
	if (isSocketCloseErrno(lastErrno)) {
	type = TTransportException.Type.NOT_OPEN;
	closeImmediately();
	}

	throw new TTransportException("Sending to socket failed: " ~
	socketErrnoString(lastErrno), type);
	}

	// send() should never return 0.
	throw new TTransportException("Sending to socket failed (0 bytes written).",
	TTransportException.Type.UNKNOWN);
	}

	/// The amount of time in which a conncetion must be established before the
	/// open() call times out.
	Duration connectTimeout = dur!"seconds"(5);

	private:
	void closeImmediately() {
	socket_.close();
	socket_ = null;
	}

	T yieldOnBlock(T)(lazy T call, TAsyncEventType eventType) {
	while (true) {
	auto result = call();
	if (result != Socket.ERROR \|\| getSocketErrno() != WOULD_BLOCK_ERRNO) return result;

	// We got an EAGAIN result, register a callback to return here once some
	// event happens and yield.

	Duration timeout = void;
	final switch (eventType) {
	case TAsyncEventType.READ:
	timeout = recvTimeout_;
	break;
	case TAsyncEventType.WRITE:
	timeout = sendTimeout_;
	break;
	}

	auto fiber = Fiber.getThis();
	assert(fiber, "Current fiber null – not running in TAsyncManager?");
	TAsyncEventReason eventReason = void;
	asyncManager_.addOneshotListener(socket_, eventType, timeout,
	scopedDelegate((TAsyncEventReason reason) {
	eventReason = reason;
	fiber.call();
	})
	);

	// Yields execution back to the async manager, will return back here once
	// the above listener is called.
	Fiber.yield();

	if (eventReason == TAsyncEventReason.TIMED_OUT) {
	// If we are cancelling the request due to a timed out operation, the
	// connection is in an undefined state, because the server could decide
	// to send the requested data later, or we could have already been half-
	// way into writing a request. Thus, we close the connection to make any
	// possibly queued up work items fail immediately. Besides, the server
	// is not very likely to immediately recover after a socket-level
	// timeout has expired anyway.
	closeImmediately();

	throw new TTransportException("Timed out while waiting for socket " ~
	"to get ready to " ~ to!string(eventType) ~ ".",
	TTransportException.Type.TIMED_OUT);
	}
	}
	}

	/// The TAsyncSocketManager to use for non-blocking I/O.
	TAsyncSocketManager asyncManager_;
	}

	private {
	// std.socket doesn't include SO_ERROR for reasons unknown.
	version (linux) {
	enum SO_ERROR = 4;
	} else version (OSX) {
	enum SO_ERROR = 0x1007;
	} else version (FreeBSD) {
	enum SO_ERROR = 0x1007;
	} else version (Win32) {
	import std.c.windows.winsock : SO_ERROR;
	} else static assert(false, "Don't know SO_ERROR on this platform.");

	// This hack forces a delegate literal to be scoped, even if it is passed to
	// a function accepting normal delegates as well. DMD likes to allocate the
	// context on the heap anyway, but it seems to work for LDC.
	import std.traits : isDelegate;
	auto scopedDelegate(D)(scope D d) if (isDelegate!D) {
	return d;
	}
	}