#include "stdafx.h"
#include "HandleStream.h"
#include "Core/Exception.h"
#include "OS/IORequest.h"
#include <limits>

namespace storm {

	// Helper to make a Duration into a ms interval, clamping as necessary.
	static nat toMs(Duration duration) {
		Long ms = duration.inMs();
		if (ms < 0)
			return 0;
		if (ms > std::numeric_limits<nat>::max())
			return std::numeric_limits<nat>::max();
		return nat(ms);
	}

	static void attach(os::Handle h, os::Thread &attached) {
		if (attached == os::Thread::invalid) {
			attached = os::Thread::current();
			attached.attach(h);
		}
	}

	/**
	 * System-specific helpers. These all behave as if the handle was blocking.
	 */

#if defined(WINDOWS)

	void close(os::Handle &h, os::Thread &attached) {
		os::Handle toClose = h;
		h = os::Handle();

		CloseHandle(toClose.v());
	}

	static PeekReadResult read(os::Handle h, os::Thread &attached, void *dest, Nat limit,
							Bool noPos, sys::ErrorCode &errorCode, Duration timeout = Duration()) {
		attach(h, attached);

		os::IORequest request(h, attached, toMs(timeout));

		LARGE_INTEGER pos;
		pos.QuadPart = 0;
		if (!noPos && SetFilePointerEx(h.v(), pos, &pos, FILE_CURRENT)) {
			// There seems to be a poblem when reading from the end of a file asynchronously.
			LARGE_INTEGER len;
			len.QuadPart = 0;
			GetFileSizeEx(h.v(), &len);
			if (pos.QuadPart >= len.QuadPart) {
				return PeekReadResult::end();
			}

			request.Offset = pos.LowPart;
			request.OffsetHigh = pos.HighPart;
		} else {
			// If we can not seek, it means that the offset should be zero.
			request.Offset = 0;
			request.OffsetHigh = 0;
		}

		int error = 0;
		// Note: When using ReadFile with sockets, it will sometimes indicate synchronous
		// completion *and* notify the IO completion port. Therefore, we will not trust the return
		// value, and assume asynchronous completion anyway.
		if (!ReadFile(h.v(), dest, DWORD(limit), NULL, &request))
			error = GetLastError();

		if (error == ERROR_IO_PENDING || error == 0) {
			// Completing async...
			request.wake.down();

			if (!noPos) {
				// Advance the file pointer.
				pos.QuadPart = request.bytes;
				SetFilePointerEx(h.v(), pos, NULL, FILE_CURRENT);
			}

			// Was the operation cancelled?
			if (request.bytes == 0 && request.error == ERROR_OPERATION_ABORTED)
				return PeekReadResult::timeout();
			// Error?
			if (request.error)
				errorCode = fromSystemError(request.error);

		} else {
			// Failed.
			request.bytes = 0;
			errorCode = fromSystemError(error);
		}

		return PeekReadResult::success(request.bytes);
	}

	static Nat write(os::Handle h, os::Thread &attached, const void *src, Nat limit,
					Bool noPos, sys::ErrorCode &errorCode, Duration timeout = Duration()) {
		attach(h, attached);

		os::IORequest request(h, attached, toMs(timeout));

		LARGE_INTEGER pos;
		pos.QuadPart = 0;
		if (!noPos && SetFilePointerEx(h.v(), pos, &pos, FILE_CURRENT)) {
			// All is well.
			request.Offset = pos.LowPart;
			request.OffsetHigh = pos.HighPart;
		} else {
			// If we can not seek, it means that the offset should be zero.
			request.Offset = 0;
			request.OffsetHigh = 0;
		}

		int error = 0;
		// Note: When using WriteFile with sockets, it will sometimes indicate synchronous
		// completion *and* notify the IO completion port. Therefore, we will not trust the return
		// value, and assume asynchronous completion anyway.
		if (!WriteFile(h.v(), src, DWORD(limit), NULL, &request))
			error = GetLastError();

		if (error == ERROR_IO_PENDING || error == 0) {
			// Completing async...
			request.wake.down();

			// Advance the file pointer.
			if (!noPos) {
				pos.QuadPart = request.bytes;
				SetFilePointerEx(h.v(), pos, NULL, FILE_CURRENT);
			}

			// Error?
			if (request.error != 0)
				errorCode = fromSystemError(request.error);

		} else {
			// Failed.
			request.bytes = 0;
			errorCode = fromSystemError(error);
		}

		return request.bytes;
	}

	static void seek(os::Handle h, Word to) {
		LARGE_INTEGER pos;
		pos.QuadPart = to;
		SetFilePointerEx(h.v(), pos, NULL, FILE_BEGIN);
	}

	static Word tell(os::Handle h) {
		LARGE_INTEGER pos;
		pos.QuadPart = 0;
		SetFilePointerEx(h.v(), pos, &pos, FILE_CURRENT);
		return pos.QuadPart;
	}

	static Word length(os::Handle h) {
		LARGE_INTEGER len;
		GetFileSizeEx(h.v(), &len);
		return len.QuadPart;
	}

	static os::Handle dupHandle(os::Handle src) {
		if (!src)
			return os::Handle();

		HANDLE dest = INVALID_HANDLE_VALUE;
		HANDLE proc = GetCurrentProcess();
		if (!DuplicateHandle(proc, src.v(), proc, &dest, DUPLICATE_SAME_ACCESS, FALSE, 0)) {
			PLN(L"Failed to duplicate handle: " << GetLastError());
			return os::Handle();
		}

		return dest;
	}

	static os::Handle openStd(DWORD id, bool input) {
		return dupHandle(GetStdHandle(id));
	}

#elif defined(LINUX_IO_URING)

	void close(os::Handle &h, os::Thread &attached) {
		if (h != os::Handle()) {
			// Since 'close' might be called from a destructor, it is not always called on the
			// proper OS thread. Since we now rarely use Semaphore to block OS threads entirely,
			// this is usually not a large problem (if we did, we could deadlock since the result to
			// our IOring request would be posted to a thread that is potentially blocked for
			// something else). Regardless, inter-thread communication is not ideal in destructors
			// regardless, so we try to avoid it.
			if (attached == os::Thread::current()) {
				// Closed from the proper thread. Detach and post a close request as a single operation.
				os::IORequest r(h, attached);
				r.request.opcode = IORING_OP_CLOSE;

				// submitAndDetach may block, so update 'h' already here so that we don't try to
				// close the same fd multiple times.
				h = os::Handle();
				r.submitAndDetach();
			} else if (attached != os::Thread::invalid) {
				// Posted from a different thread. Do not block this thread by heap-allocating the
				// request and asking it to deallocate itself once completed.
				os::IORequest *r = new os::IORequest(h, attached);
				r->request.opcode = IORING_OP_CLOSE;

				h = os::Handle();
				r->submitDetachDelete();
			} else {
				// This is not likely to happen, just close the fd normally as a last resort.
				::close(h.v());
			}
		}

		attached = os::Thread::invalid;
		h = os::Handle();
	}

	static PeekReadResult read(os::Handle h, os::Thread &attached, void *dest, Nat limit,
							Bool noPos, sys::ErrorCode &errorCode, Duration timeout = Duration()) {
		attach(h, attached);

		os::IORequest r(h, attached);

		r.request.opcode = IORING_OP_READ;
		r.request.addr = reinterpret_cast<size_t>(dest);
		r.request.len = limit;
		r.request.off = -1; // Advances the file pointer as usual.

		int result = r.submit(toMs(timeout));
		// Note: ECANCELED means that the timeout was reached, or that the file was closed during
		// the operation (since we cancel it then). We don't report that as an error, since none of
		// the other implementations do.
		if (result < 0 && result != -ECANCELED)
			errorCode = fromSystemError(-result);

		if (result <= 0)
			return r.timeout ? PeekReadResult::timeout() : PeekReadResult::end();
		else
			return PeekReadResult::success(Nat(result));
	}

	static Nat write(os::Handle h, os::Thread &attached, const void *src, Nat limit,
					Bool noPos, sys::ErrorCode &errorCode, Duration timeout = Duration()) {
		attach(h, attached);

		os::IORequest r(h, attached);

		r.request.opcode = IORING_OP_WRITE;
		r.request.addr = reinterpret_cast<size_t>(src);
		r.request.len = limit;
		r.request.off = -1; // Advances the file pointer as usual.

		int result = r.submit(toMs(timeout));
		if (result < 0 && result != -ECANCELED)
			errorCode = fromSystemError(-result);

		if (result <= 0)
			return 0;
		else
			return Nat(result);
	}

	static void seek(os::Handle h, Word to) {
		lseek64(h.v(), to, SEEK_SET);
	}

	static Word tell(os::Handle h) {
		off64_t r = lseek64(h.v(), 0, SEEK_CUR);
		if (r < 0)
			return 0;
		return Word(r);
	}

	static Word length(os::Handle h) {
		off64_t old = lseek64(h.v(), 0, SEEK_CUR);
		if (old < 0)
			return 0;

		off64_t size = lseek64(h.v(), 0, SEEK_END);
		lseek64(h.v(), old, SEEK_SET);
		return Word(size);
	}

	static os::Handle dupHandle(os::Handle src) {
		return os::Handle(dup(src.v()));
	}


#elif defined(POSIX)

	void close(os::Handle &h, os::Thread &attached) {
		if (attached != os::Thread::invalid)
			attached.detach(h);

		::close(h.v());
		attached = os::Thread::invalid;
		h = os::Handle();
	}

	struct WaitResult {
		bool closed;
		bool timeout;
	};

	// Returns 'false' if the handle was closed (by us) during the operation.
	static WaitResult doWait(os::Handle h, os::Thread &attached, os::IORequest::Type type, Duration timeout) {
		attach(h, attached);

		os::IORequest request(h, type, attached, toMs(timeout));
		request.wake.wait();
		WaitResult r = {
			request.closed,
			request.timeout
		};
		return r;
	}

	static PeekReadResult read(os::Handle h, os::Thread &attached, void *dest, Nat limit,
							Bool noPos, sys::ErrorCode &errorCode, Duration timeout = Duration()) {
		while (true) {
			ssize_t r = ::read(h.v(), dest, size_t(limit));
			if (r >= 0)
				return PeekReadResult::success(Nat(r));

			int error = errno;
			if (error == EINTR) {
				// Aborted by a signal. Retry.
				continue;
			} else if (error == EAGAIN) {
				// Wait for more data.
				WaitResult r = doWait(h, attached, os::IORequest::read, timeout);
				if (r.timeout)
					return PeekReadResult::timeout();
				if (r.closed)
					break;
			} else {
				// Other error.
				errorCode = fromSystemError(error);
				break;
			}
		}

		return PeekReadResult::end();
	}

	static Nat write(os::Handle h, os::Thread &attached, const void *src, Nat limit,
					Bool noPos, sys::ErrorCode &errorCode, Duration timeout = Duration()) {
		while (true) {
			ssize_t r = ::write(h.v(), src, size_t(limit));
			if (r >= 0)
				return Nat(r);

			int error = errno;
			if (error == EINTR) {
				// Aborted by a signal. Retry.
				continue;
			} else if (error == EAGAIN) {
				// Wait for more data.
				WaitResult r = doWait(h, attached, os::IORequest::write, timeout);
				if (r.closed || r.timeout)
					break;
			} else {
				// Other error.
				errorCode = fromSystemError(error);
				break;
			}
		}

		return 0;
	}

	static void seek(os::Handle h, Word to) {
		lseek64(h.v(), to, SEEK_SET);
	}

	static Word tell(os::Handle h) {
		off64_t r = lseek64(h.v(), 0, SEEK_CUR);
		if (r < 0)
			return 0;
		return Word(r);
	}

	static Word length(os::Handle h) {
		off64_t old = lseek64(h.v(), 0, SEEK_CUR);
		if (old < 0)
			return 0;

		off64_t size = lseek64(h.v(), 0, SEEK_END);
		lseek64(h.v(), old, SEEK_SET);
		return Word(size);
	}

	static os::Handle dupHandle(os::Handle src) {
		return os::Handle(dup(src.v()));
	}

#else
#error "Please implement file IO for your OS."
#endif

	/**
	 * Regular input stream.
	 */

	HandleIStream::HandleIStream(os::Handle h)
		: handle(h),
		  attachedTo(os::Thread::invalid),
		  noClose(false),
		  currError(sys::none) {}

	HandleIStream::HandleIStream(os::Handle h, os::Thread t)
		: handle(h),
		  attachedTo(t),
		  noClose(false),
		  currError(sys::none) {}

	// HandleIStream::HandleIStream(const HandleIStream &o)
	// 	: PeekIStream(o),
	// 	  handle(dupHandle(o.handle)),
	// 	  attachedTo(os::Thread::invalid) {}

	HandleIStream::HandleIStream(const HandleIStream &o)
		: PeekIStream(o), attachedTo(os::Thread::invalid), noClose(o.noClose), currError(o.currError) {
		throw new (this) NotSupported(S("Copying HandleIStream"));
	}

	HandleIStream::~HandleIStream() {
		if (handle)
			storm::close(handle, attachedTo);
	}

	Bool HandleIStream::more() {
		if (!handle)
			return false;

		return PeekIStream::more();
	}

	void HandleIStream::close() {
		if (handle && !noClose)
			storm::close(handle, attachedTo);
		handle = os::Handle();

		PeekIStream::close();
	}

	PeekReadResult HandleIStream::doRead(byte *to, Nat count) {
		if (handle)
			return storm::read(handle, attachedTo, to, count, noPos, currError);
		else
			return PeekReadResult::end();
	}

	/**
	 * Random access stream.
	 */

	HandleRIStream::HandleRIStream(os::Handle h)
		: handle(h),
		  attachedTo(os::Thread::invalid),
		  noClose(false),
		  currError(sys::none) {}

	HandleRIStream::HandleRIStream(os::Handle h, os::Thread t)
		: handle(h),
		  attachedTo(t),
		  noClose(false),
		  currError(sys::none) {}

	// HandleRIStream::HandleRIStream(const HandleRIStream &o)
	// 	: handle(dupHandle(o.handle)),
	// 	  attachedTo(os::Thread::invalid) {}

	HandleRIStream::HandleRIStream(const HandleRIStream &o) : attachedTo(os::Thread::invalid), noClose(false), currError(o.currError) {
		throw new (this) NotSupported(S("Copying HandleIStream"));
	}

	HandleRIStream::~HandleRIStream() {
		if (handle)
			storm::close(handle, attachedTo);
	}

	void HandleRIStream::deepCopy(CloneEnv *e) {
		// Nothing needs to be done.
	}

	Bool HandleRIStream::more() {
		if (!handle)
			return false;

		return tell() < length();
	}

	Buffer HandleRIStream::read(Buffer b) {
		Nat start = b.filled();

		if (!handle)
			return b;

		if (start >= b.count())
			return b;

		PeekReadResult r = storm::read(handle, attachedTo, b.dataPtr() + start, b.count() - start, noPos, currError);
		b.filled(r.bytesRead() + start);
		return b;
	}

	Buffer HandleRIStream::peek(Buffer b) {
		if (!handle)
			return b;

		Word pos = tell();
		b = read(b);
		seek(pos);

		return b;
	}

	void HandleRIStream::close() {
		if (handle && !noClose)
			storm::close(handle, attachedTo);
		handle = os::Handle();
	}

	RIStream *HandleRIStream::randomAccess() {
		return this;
	}

	void HandleRIStream::seek(Word to) {
		storm::seek(handle, to);
	}

	Word HandleRIStream::tell() {
		if (!handle)
			return 0;

		return storm::tell(handle);
	}

	Word HandleRIStream::length() {
		if (!handle)
			return 0;

		return storm::length(handle);
	}

	/**
	 * With timeout.
	 */

	HandleTimeoutIStream::HandleTimeoutIStream(os::Handle handle) : HandleIStream(handle), timeout() {}

	HandleTimeoutIStream::HandleTimeoutIStream(os::Handle handle, os::Thread attachedTo) : HandleIStream(handle), timeout() {}

	PeekReadResult HandleTimeoutIStream::doRead(byte *to, Nat count) {
		if (handle)
			return storm::read(handle, attachedTo, to, count, noPos, currError, timeout);
		else
			return PeekReadResult::end();
	}

	/**
	 * Output stream.
	 */

	HandleOStream::HandleOStream(os::Handle h)
		: handle(h),
		  attachedTo(os::Thread::invalid),
		  noClose(false),
		  currError(sys::none) {}

	HandleOStream::HandleOStream(os::Handle h, os::Thread t)
		: handle(h),
		  attachedTo(t),
		  noClose(false),
		  currError(sys::none) {}

	// HandleOStream::HandleOStream(const HandleOStream &o)
	// 	: handle(dupHandle(o.handle)),
	// 	  attachedTo(os::Thread::invalid) {}

	HandleOStream::HandleOStream(const HandleOStream &o)
		: attachedTo(os::Thread::invalid), noClose(o.noClose), currError(o.currError) {
		throw new (this) NotSupported(S("Copying HandleIStream"));
	}

	HandleOStream::~HandleOStream() {
		if (handle)
			storm::close(handle, attachedTo);
	}

	Nat HandleOStream::write(Buffer to, Nat start) {
		Nat consumed = 0;
		start = min(start, to.filled());
		if (handle) {
			while (start < to.filled()) {
				Nat r = storm::write(handle, attachedTo, to.dataPtr() + start, to.filled() - start, noPos, currError);
				if (r == 0)
					break;
				start += r;
				consumed += r;
			}
		}
		return consumed;
	}

	void HandleOStream::close() {
		if (handle && !noClose)
			storm::close(handle, attachedTo);
		handle = os::Handle();
	}

	/**
	 * With timeout.
	 */

	HandleTimeoutOStream::HandleTimeoutOStream(os::Handle h) : HandleOStream(h), timeout() {}

	HandleTimeoutOStream::HandleTimeoutOStream(os::Handle h, os::Thread attachedTo)
		: HandleOStream(h, attachedTo), timeout() {}

	Nat HandleTimeoutOStream::write(Buffer to, Nat start) {
		Nat consumed = 0;
		start = min(start, to.filled());
		if (handle) {
			while (start < to.filled()) {
				Nat r = storm::write(handle, attachedTo, to.dataPtr() + start, to.filled() - start, noPos, currError, timeout);
				if (r == 0)
					break;
				start += r;
				consumed += r;
			}
		}
		return consumed;
	}

}