Experimental

2025-12-06 00:13:33 +00:00 · 2019-04-26 16:07:06 -07:00
parent 18e4dd4548
commit 1e58eaf103
3 changed files with 344 additions and 11 deletions
--- a/microscript/ILibDuktape_net.c
+++ b/microscript/ILibDuktape_net.c
@@ -21,10 +21,12 @@ limitations under the License.
 #include "ILibDuktape_DuplexStream.h"
 #include "ILibDuktapeModSearch.h"
 #include "ILibDuktape_EventEmitter.h"
+#include "ILibDuktape_ChildProcess.h"
 #include "microstack/ILibAsyncSocket.h"
 #include "microstack/ILibCrypto.h"
 #include "microstack/ILibAsyncServerSocket.h"
 #include "microstack/ILibRemoteLogging.h"
+#include "microstack/ILibProcessPipe.h"

 #ifdef _POSIX
 #include <sys/types.h>
@@ -32,6 +34,11 @@ limitations under the License.
 #include <sys/un.h>
 #endif

+#ifdef WIN32
+#include <accctrl.h>
+#include <AclAPI.h>
+#endif
+
 typedef struct ILibDuktape_net_socket
 {
 	duk_context *ctx;
@@ -71,6 +78,28 @@ typedef struct ILibDuktape_net_server_session
 int ILibDuktape_TLS_ctx2socket = -1;
 int ILibDuktape_TLS_ctx2server = -1;

+#ifdef WIN32
+#define ILibDuktape_net_IPC_BUFFERSIZE	4096
+typedef struct ILibDuktape_net_WindowsIPC
+{
+	ILibProcessPipe_Manager manager;
+	duk_context *ctx;
+	void *object, *mSocket;
+	HANDLE mPipeHandle;
+	ILibProcessPipe_Pipe mPipe;
+
+	OVERLAPPED overlapped;
+	ILibDuktape_DuplexStream *ds;
+
+	int processingRead;
+	char *buffer;
+	int bufferLength;
+	int bufferOffset;
+	int bytesLeft;
+
+}ILibDuktape_net_WindowsIPC;
+#endif
+
 #define ILibDuktape_SecureContext2CertBuffer	"\xFF_SecureContext2CertBuffer"
 #define ILibDuktape_SecureContext2SSLCTXPTR		"\xFF_SecureContext2SSLCTXPTR"
 #define ILibDuktape_GlobalTunnel_DataPtr		"\xFF_GlobalTunnel_DataPtr"
@@ -78,6 +107,8 @@ int ILibDuktape_TLS_ctx2server = -1;
 #define ILibDuktape_net_Server_buffer			"\xFF_FixedBuffer"
 #define ILibDuktape_net_Server_Session_buffer	"\xFF_SessionFixedBuffer"
 #define ILibDuktape_net_socket_ptr				"\xFF_SocketPtr"
+#define ILibDuktape_net_WindowsIPC_Buffer		"\xFF_WindowsIPC"
+#define ILibDuktape_net_WindowsIPC_PendingArray "\xFF_WindowsIPC_PendingArray"
 #define ILibDuktape_SERVER2ContextTable			"\xFF_Server2ContextTable"
 #define ILibDuktape_SERVER2OPTIONS				"\xFF_ServerToOptions"
 #define ILibDuktape_SERVER2LISTENOPTIONS		"\xFF_ServerToListenOptions"
@@ -689,6 +720,161 @@ void ILibDuktape_net_server_OnSendOK(ILibAsyncServerSocket_ServerModule AsyncSer

 	ILibDuktape_DuplexStream_Ready(session->stream);
 }
+
+#ifdef WIN32
+extern void ILibProcessPipe_FreePipe(ILibProcessPipe_Pipe pipeObject);
+
+int ILibDuktape_net_server_IPC_unshiftSink(ILibDuktape_DuplexStream *sender, int unshiftBytes, void *user)
+{
+	ILibDuktape_net_WindowsIPC *winIPC = (ILibDuktape_net_WindowsIPC*)user;
+	if (!ILibMemory_CanaryOK(user)) { return(0); }
+
+	winIPC->bufferOffset += (winIPC->bytesLeft - unshiftBytes);
+	winIPC->bytesLeft = unshiftBytes;
+	return(unshiftBytes);
+}
+void ILibDuktape_net_server_IPC_readsink(ILibProcessPipe_Pipe sender, void *user, DWORD dwErrorCode, char *buffer, int bufferLen)
+{
+	ILibDuktape_net_WindowsIPC *winIPC = (ILibDuktape_net_WindowsIPC*)user;
+	if (!ILibMemory_CanaryOK(user)) { return; }
+
+	if (dwErrorCode == 0)
+	{
+		ILibDuktape_DuplexStream_WriteData(winIPC->ds, buffer, bufferLen);
+	}
+	else
+	{
+		ILibDuktape_DuplexStream_Closed(winIPC->ds);
+	}
+}
+void ILibDuktape_net_server_IPC_PauseSink(ILibDuktape_DuplexStream *sender, void *user)
+{
+	// No-OP, becuase all we need to so is set Paused flag, which is already the  case when we get here
+}
+void ILibDuktape_net_server_IPC_ResumeSink(ILibDuktape_DuplexStream *sender, void *user)
+{
+	ILibDuktape_net_WindowsIPC *winIPC = (ILibDuktape_net_WindowsIPC*)user;
+	if (winIPC->processingRead != 0) { return; }
+	winIPC->processingRead = 1;
+
+	if (winIPC->buffer == NULL)
+	{
+		winIPC->buffer = ILibMemory_Allocate(ILibDuktape_net_IPC_BUFFERSIZE, 0, NULL, NULL);
+		winIPC->bufferLength = ILibDuktape_net_IPC_BUFFERSIZE;
+		winIPC->bufferOffset = 0;
+		winIPC->bytesLeft = 0;
+	}
+
+	if (winIPC->bytesLeft <= 0)
+	{
+		winIPC->bytesLeft = 0;
+		winIPC->bufferOffset = 0;
+		ILibProcessPipe_Pipe_ReadEx(winIPC->mPipe, winIPC->buffer, winIPC->bufferLength, winIPC, ILibDuktape_net_server_IPC_readsink);
+	}
+	else
+	{
+		// Check to see if we can drain any of the buffer first
+		int tmpBytesLeft;
+		while (winIPC->ds->readableStream->paused == 0)
+		{
+			tmpBytesLeft = winIPC->bytesLeft;
+			ILibDuktape_DuplexStream_WriteData(winIPC->ds, winIPC->buffer + winIPC->bufferOffset, winIPC->bytesLeft);
+			if (tmpBytesLeft == winIPC->bytesLeft) { break; } // No Data was consumed
+			if (winIPC->bytesLeft <= 0)
+			{
+				winIPC->bytesLeft = 0;
+				winIPC->bufferOffset = 0;
+			}
+			if (winIPC->ds->readableStream->paused == 0 && (winIPC->bytesLeft == 0 || (winIPC->bytesLeft > 0 && tmpBytesLeft == winIPC->bytesLeft)))
+			{
+				ILibProcessPipe_Pipe_ReadEx(winIPC->mPipe, winIPC->buffer + winIPC->bufferOffset + winIPC->bytesLeft, winIPC->bufferLength - winIPC->bufferOffset - winIPC->bytesLeft, winIPC, ILibDuktape_net_server_IPC_readsink);
+				break;
+			}
+		}
+	}
+	winIPC->processingRead = 0;
+}
+void ILibDuktape_net_server_IPC_WriteCompletionEvent(ILibProcessPipe_Pipe sender, void *user, DWORD errorCode, int bytesWritten)
+{
+	if (!ILibMemory_CanaryOK(user)) { return; }
+	ILibDuktape_net_WindowsIPC *winIPC = (ILibDuktape_net_WindowsIPC*)user;
+	duk_idx_t top = duk_get_top(winIPC->ctx);
+
+	duk_push_heapptr(winIPC->ctx, winIPC->object);									// [obj]
+	duk_get_prop_string(winIPC->ctx, -1, ILibDuktape_net_WindowsIPC_PendingArray);	// [obj][array]
+	duk_get_prop_string(winIPC->ctx, -1, "shift");									// [obj][array][shift]
+	duk_dup(winIPC->ctx, -2);														// [obj][array][shift][this]
+	if (duk_pcall_method(winIPC->ctx, 0) != 0)										// [obj][array][buffer]
+	{
+		ILibDuktape_Process_UncaughtExceptionEx(winIPC->ctx, "Internal Error: net.socket.ipc.writeCompletionEvent");
+		duk_set_top(winIPC->ctx, top);												// ...
+		return;
+	}
+	duk_pop(winIPC->ctx);															// [obj][array]
+	if (duk_get_length(winIPC->ctx, -1) > 0)
+	{
+		// Still pending Writes
+		duk_get_prop_index(winIPC->ctx, -1, 0);										// [obj][array][buffer]
+		duk_size_t bufLen;
+		char *buf = (char*)Duktape_GetBuffer(winIPC->ctx, -1, &bufLen);
+		duk_set_top(winIPC->ctx, top);												// ...
+		ILibProcessPipe_Pipe_WriteEx(winIPC->mPipe, buf, (int)bufLen, winIPC, ILibDuktape_net_server_IPC_WriteCompletionEvent);
+	}
+}
+ILibTransport_DoneState ILibDuktape_net_server_IPC_WriteSink(ILibDuktape_DuplexStream *stream, char *buffer, int bufferLen, void *user)
+{
+	if (!ILibMemory_CanaryOK(user)) { return(ILibTransport_DoneState_ERROR); }
+	ILibDuktape_net_WindowsIPC *winIPC = (ILibDuktape_net_WindowsIPC*)user;
+	
+	duk_push_heapptr(winIPC->ctx, winIPC->object);											// [obj]
+	duk_get_prop_string(winIPC->ctx, -1, ILibDuktape_net_WindowsIPC_PendingArray);			// [obj][array]
+	
+	
+	char *q = duk_push_fixed_buffer(winIPC->ctx, bufferLen);								// [obj][array][buffer]
+	duk_size_t len = duk_get_length(winIPC->ctx, -2);
+	duk_put_prop_index(winIPC->ctx, -2, (duk_uarridx_t)len);								// [obj][array]
+	memcpy_s(q, bufferLen, buffer, bufferLen);
+	duk_pop_2(winIPC->ctx);																	// ...
+
+	if (len == 0)
+	{
+		// No Pending Writes
+		ILibProcessPipe_Pipe_WriteEx(winIPC->mPipe, q, bufferLen, winIPC, ILibDuktape_net_server_IPC_WriteCompletionEvent);
+	}
+
+	return(ILibTransport_DoneState_INCOMPLETE);
+}
+void ILibDuktape_net_server_IPC_EndSink(ILibDuktape_DuplexStream *stream, void *user)
+{
+	if (!ILibMemory_CanaryOK(user)) { return; }
+	ILibDuktape_net_WindowsIPC *winIPC = (ILibDuktape_net_WindowsIPC*)user;
+
+	ILibProcessPipe_FreePipe(winIPC->mPipe);
+	winIPC->mPipe = NULL;
+}
+BOOL ILibDuktape_net_server_IPC_ConnectSink(HANDLE event, ILibWaitHandle_ErrorStatus status, void* user)
+{
+	if (ILibMemory_CanaryOK(user))
+	{
+		ILibDuktape_net_WindowsIPC *winIPC = (ILibDuktape_net_WindowsIPC*)user;
+		ILibDuktape_EventEmitter_SetupEmit(winIPC->ctx, winIPC->object, "connection");	// [emit][this][connection]
+		duk_push_object(winIPC->ctx);													// [emit][this][connection][socket]
+		ILibDuktape_WriteID(winIPC->ctx, "net.socket.ipc");
+		duk_push_array(winIPC->ctx); duk_put_prop_string(winIPC->ctx, -2, ILibDuktape_net_WindowsIPC_PendingArray);
+		winIPC->mSocket = duk_get_heapptr(winIPC->ctx, -1);
+		winIPC->ds = ILibDuktape_DuplexStream_InitEx(winIPC->ctx, ILibDuktape_net_server_IPC_WriteSink, ILibDuktape_net_server_IPC_EndSink, ILibDuktape_net_server_IPC_PauseSink, ILibDuktape_net_server_IPC_ResumeSink, ILibDuktape_net_server_IPC_unshiftSink, winIPC);
+		winIPC->mPipe = ILibProcessPipe_Pipe_CreateFromExisting(winIPC->manager, winIPC->mPipeHandle, ILibProcessPipe_Pipe_ReaderHandleType_Overlapped);
+
+		if (duk_pcall_method(winIPC->ctx, 2) != 0)
+		{
+			ILibDuktape_Process_UncaughtExceptionEx(winIPC->ctx, "Error emitting net.socket.ipc.connection");
+		}
+		duk_pop(winIPC->ctx);
+	}
+	return(FALSE);
+}
+#endif
+
 duk_ret_t ILibDuktape_net_server_listen(duk_context *ctx)
 {
 	int nargs = duk_get_top(ctx);
@@ -762,15 +948,75 @@ duk_ret_t ILibDuktape_net_server_listen(duk_context *ctx)
 		}
 	}
 	
-	if (ipc != NULL)
+	if (ipc != NULL && port == 0)
 	{
-#ifdef _POSIX
+#if defined(_POSIX)
 		if (ipcLen > sizeof(ipcaddr.sun_path)) { return(ILibDuktape_Error(ctx, "Path too long")); }
 		ipcaddr.sun_family = AF_UNIX;
 		strcpy_s((char*)(ipcaddr.sun_path), sizeof(ipcaddr.sun_path), ipc);
 		server->server = ILibCreateAsyncServerSocketModuleWithMemoryEx(Duktape_GetChain(ctx), maxConnections, initalBufferSize, (struct sockaddr*)&ipcaddr,
 			ILibDuktape_net_server_OnConnect, ILibDuktape_net_server_OnDisconnect, ILibDuktape_net_server_OnReceive,
 			ILibDuktape_net_server_OnInterrupt, ILibDuktape_net_server_OnSendOK, sizeof(void*), sizeof(void*));
+#elif defined(WIN32)
+		// IPC on Windows Implemented as Named Pipe
+
+		SECURITY_ATTRIBUTES IPC_SA = { 0 };
+		SECURITY_ATTRIBUTES *pIPC_SA = &IPC_SA;
+		PACL IPC_ACL;
+		SECURITY_DESCRIPTOR IPC_SD;
+		EXPLICIT_ACCESS IPC_EA = { 0 };
+				
+		duk_push_this(ctx);
+		ILibDuktape_net_WindowsIPC *winIPC = (ILibDuktape_net_WindowsIPC*)Duktape_PushBuffer(ctx, sizeof(ILibDuktape_net_WindowsIPC));
+		duk_put_prop_string(ctx, -2, ILibDuktape_net_WindowsIPC_Buffer);
+		winIPC->overlapped.hEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
+		
+		duk_eval_string(ctx, "require('child_process');");
+		winIPC->manager = (ILibProcessPipe_Manager)Duktape_GetPointerProperty(ctx, -1, ILibDuktape_ChildProcess_Manager);
+		duk_pop(ctx);
+
+		if (Duktape_GetBooleanProperty(ctx, 0, "writableAll", 0) != 0)
+		{
+			// World Writable, so we need to set the Security Descriptor to reflect that
+			IPC_EA.grfAccessMode = SET_ACCESS;
+			IPC_EA.grfInheritance = NO_INHERITANCE;
+			IPC_EA.grfAccessPermissions = FILE_GENERIC_READ | FILE_WRITE_DATA;
+			IPC_EA.Trustee.TrusteeForm = TRUSTEE_IS_NAME;
+			IPC_EA.Trustee.TrusteeType = TRUSTEE_IS_GROUP;
+			IPC_EA.Trustee.ptstrName = "EVERYONE";
+
+			SetEntriesInAcl(1, &IPC_EA, NULL, &IPC_ACL);
+			InitializeSecurityDescriptor(&IPC_SD, SECURITY_DESCRIPTOR_REVISION);
+			SetSecurityDescriptorDacl(&IPC_SD, TRUE, IPC_ACL, FALSE);
+
+			memset(&IPC_SA, 0, sizeof(SECURITY_ATTRIBUTES));
+			IPC_SA.nLength = sizeof(SECURITY_ATTRIBUTES);
+			IPC_SA.bInheritHandle = FALSE;
+			IPC_SA.lpSecurityDescriptor = &IPC_SD;
+		}
+		else
+		{
+			// Default security is Read/Write for LocalSystem and owner, and Read for everybody else
+			pIPC_SA = NULL;
+		}
+
+		winIPC->mPipeHandle = CreateNamedPipeA((LPCSTR)ipc, PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED,
+			PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_REJECT_REMOTE_CLIENTS,
+			1, ILibDuktape_net_IPC_BUFFERSIZE, ILibDuktape_net_IPC_BUFFERSIZE, 0, pIPC_SA);
+		if (winIPC->mPipeHandle == INVALID_HANDLE_VALUE)
+		{
+			CloseHandle(winIPC->overlapped.hEvent);
+			duk_del_prop_string(ctx, -1, ILibDuktape_net_WindowsIPC_Buffer);
+			return(ILibDuktape_Error(ctx, "Error Creating Named Pipe: %s", ipc));
+		}
+		winIPC->ctx = ctx;
+		winIPC->object = duk_get_heapptr(ctx, -1);
+		
+		ConnectNamedPipe(winIPC->mPipeHandle, &winIPC->overlapped);
+		ILibProcessPipe_WaitHandle_Add2(winIPC->manager, winIPC->overlapped.hEvent, winIPC, ILibDuktape_net_server_IPC_ConnectSink);
+
+		if (pIPC_SA != NULL) { LocalFree(IPC_ACL); }
+		return(1);
 #endif
 	}
 	else
@@ -828,7 +1074,8 @@ duk_ret_t ILibDuktape_net_server_listen(duk_context *ctx)
 	ignore_result(backlog);
 #endif

-	return 0;
+	duk_push_this(ctx);
+	return 1;
 }
 duk_ret_t ILibDuktape_net_server_Finalizer(duk_context *ctx)
 {
--- a/microstack/ILibProcessPipe.c
+++ b/microstack/ILibProcessPipe.c
@@ -82,8 +82,8 @@ typedef struct ILibProcessPipe_PipeObject
 	HANDLE mPipe_Reader_ResumeEvent;
 	HANDLE mPipe_ReadEnd;
 	HANDLE mPipe_WriteEnd;
-	struct _OVERLAPPED *mOverlapped;
-	void *mOverlapped_opaqueData;
+	OVERLAPPED *mOverlapped,*mwOverlapped;
+	void *mOverlapped_opaqueData, *user3, *user4;
 #else
 	int mPipe_ReadEnd, mPipe_WriteEnd;
 #endif
@@ -163,6 +163,14 @@ typedef struct ILibProcessPipe_WaitHandle
 	int timeRemaining;
 	int timeout;
 }ILibProcessPipe_WaitHandle;
+typedef struct ILibProcessPipe_WaitHandle_APC
+{
+	HANDLE callingThread;
+	HANDLE ev;
+	ILibWaitHandle_ErrorStatus status;
+	ILibProcessPipe_WaitHandle_Handler callback;
+	void *user;
+}ILibProcessPipe_WaitHandle_APC;
 HANDLE ILibProcessPipe_Manager_GetWorkerThread(ILibProcessPipe_Manager mgr)
 {
 	return(((ILibProcessPipe_Manager_Object*)mgr)->workerThread);
@@ -236,6 +244,32 @@ void ILibProcessPipe_WaitHandle_Add_WithNonZeroTimeout(ILibProcessPipe_Manager m
 	ILibProcessPipe_WaitHandle_AddEx(mgr, waitHandle);
 }

+void __stdcall ILibProcessPipe_WaitHandle_Add2_apcsink(ULONG_PTR obj)
+{
+	if (ILibMemory_CanaryOK((void*)obj))
+	{
+		ILibProcessPipe_WaitHandle_APC *apcState = (ILibProcessPipe_WaitHandle_APC*)obj;
+		if (apcState->callback != NULL) { apcState->callback(apcState->ev, apcState->status, apcState->user); }
+		ILibMemory_Free(apcState);
+	}
+}
+BOOL ILibProcessPipe_WaitHandle_Add2_sink(HANDLE event, ILibWaitHandle_ErrorStatus status, void* user)
+{
+	if (ILibMemory_CanaryOK(user))
+	{
+		QueueUserAPC((PAPCFUNC)ILibProcessPipe_WaitHandle_Add2_apcsink, ((ILibProcessPipe_WaitHandle_APC*)user)->callingThread, (ULONG_PTR)user);
+	}
+	return(FALSE);
+}
+void ILibProcessPipe_WaitHandle_Add2_WithNonZeroTimeout(ILibProcessPipe_Manager mgr, HANDLE event, int milliseconds, void *user, ILibProcessPipe_WaitHandle_Handler callback)
+{
+	ILibProcessPipe_WaitHandle_APC *apcState = ILibMemory_SmartAllocate(sizeof(ILibProcessPipe_WaitHandle_APC));
+	DuplicateHandle(GetCurrentProcess(), GetCurrentThread(), GetCurrentProcess(), &apcState->callingThread, THREAD_SET_CONTEXT, FALSE, 0);
+	apcState->callback = callback;
+	apcState->user = user;
+	ILibProcessPipe_WaitHandle_Add_WithNonZeroTimeout(mgr, event, milliseconds, apcState, ILibProcessPipe_WaitHandle_Add2_sink);
+}
+
 void ILibProcessPipe_Manager_WindowsRunLoopEx(void *arg)
 {
 	ILibProcessPipe_Manager_Object *manager = (ILibProcessPipe_Manager_Object*)arg;
@@ -518,7 +552,7 @@ void ILibProcessPipe_FreePipe(ILibProcessPipe_PipeObject *pipeObject)

 #ifdef WIN32
 	if (pipeObject->mPipe_ReadEnd != NULL) { CloseHandle(pipeObject->mPipe_ReadEnd); }
-	if (pipeObject->mPipe_WriteEnd != NULL) { CloseHandle(pipeObject->mPipe_WriteEnd); }
+	if (pipeObject->mPipe_WriteEnd != NULL && pipeObject->mPipe_WriteEnd != pipeObject->mPipe_ReadEnd) { CloseHandle(pipeObject->mPipe_WriteEnd); }
 	if (pipeObject->mOverlapped != NULL) { CloseHandle(pipeObject->mOverlapped->hEvent); free(pipeObject->mOverlapped); }
 	if (pipeObject->mPipe_Reader_ResumeEvent != NULL) { CloseHandle(pipeObject->mPipe_Reader_ResumeEvent); }
 #endif
@@ -565,9 +599,10 @@ ILibProcessPipe_Pipe ILibProcessPipe_Pipe_CreateFromExistingWithExtraMemory(ILib
 #ifdef WIN32
 	if (handleType == ILibProcessPipe_Pipe_ReaderHandleType_Overlapped)
 	{
-		if ((retVal->mOverlapped = (struct _OVERLAPPED*)malloc(sizeof(struct _OVERLAPPED))) == NULL) { ILIBCRITICALEXIT(254); }
-		memset(retVal->mOverlapped, 0, sizeof(struct _OVERLAPPED));
+		void *tmpExtra;
+		retVal->mOverlapped = (OVERLAPPED*)ILibMemory_Allocate(sizeof(OVERLAPPED), sizeof(void*), NULL, &tmpExtra);
 		if ((retVal->mOverlapped->hEvent = CreateEvent(NULL, TRUE, FALSE, NULL)) == NULL) { ILIBCRITICALEXIT(254); }
+		((void**)tmpExtra)[0] = retVal;
 	}
 #else
 	fcntl(existingPipe, F_SETFL, O_NONBLOCK);
@@ -1555,16 +1590,22 @@ ILibTransport_DoneState ILibProcessPipe_Pipe_Write(ILibProcessPipe_Pipe po, char
 			}
 			else
 			{
+				if (pipeObject->manager != NULL)
+				{
 #ifdef WIN32
-				ILibRemoteLogging_printf(ILibChainGetLogger(pipeObject->manager->ChainLink.ParentChain), ILibRemoteLogging_Modules_Microstack_Pipe, ILibRemoteLogging_Flags_VerbosityLevel_1, "ILibProcessPipe[Write]: BrokenPipe(%d) on Pipe: %p", GetLastError(), (void*)pipeObject);
+					ILibRemoteLogging_printf(ILibChainGetLogger(pipeObject->manager->ChainLink.ParentChain), ILibRemoteLogging_Modules_Microstack_Pipe, ILibRemoteLogging_Flags_VerbosityLevel_1, "ILibProcessPipe[Write]: BrokenPipe(%d) on Pipe: %p", GetLastError(), (void*)pipeObject);
 #else
-				ILibRemoteLogging_printf(ILibChainGetLogger(pipeObject->manager->ChainLink.ParentChain), ILibRemoteLogging_Modules_Microstack_Pipe, ILibRemoteLogging_Flags_VerbosityLevel_1, "ILibProcessPipe[Write]: BrokenPipe(%d) on Pipe: %p", result < 0 ? errno : 0, (void*)pipeObject);
+					ILibRemoteLogging_printf(ILibChainGetLogger(pipeObject->manager->ChainLink.ParentChain), ILibRemoteLogging_Modules_Microstack_Pipe, ILibRemoteLogging_Flags_VerbosityLevel_1, "ILibProcessPipe[Write]: BrokenPipe(%d) on Pipe: %p", result < 0 ? errno : 0, (void*)pipeObject);
 #endif
+				}
 				ILibQueue_UnLock(pipeObject->WriteBuffer);
 				if (pipeObject->brokenPipeHandler != NULL)
 				{
 #ifdef WIN32
-					ILibProcessPipe_WaitHandle_Remove(pipeObject->manager, pipeObject->mOverlapped->hEvent); // Pipe Broken, so remove ourselves from the processing loop
+					if (pipeObject->manager != NULL)
+					{
+						ILibProcessPipe_WaitHandle_Remove(pipeObject->manager, pipeObject->mOverlapped->hEvent); // Pipe Broken, so remove ourselves from the processing loop
+					}
 #endif
 					((ILibProcessPipe_GenericBrokenPipeHandler)pipeObject->brokenPipeHandler)(pipeObject);
 				}
@@ -1602,6 +1643,44 @@ void ILibProcessPipe_Pipe_AddPipeReadHandler(ILibProcessPipe_Pipe targetPipe, in
 	ILibProcessPipe_Process_StartPipeReader(targetPipe, bufferSize, &ILibProcessPipe_Pipe_ReadSink, targetPipe, OnReadHandler);
 }
 #ifdef WIN32
+void __stdcall ILibProcessPipe_Pipe_Read_CompletionRoutine(DWORD dwErrorCode, DWORD dwNumberOfBytesTransfered, LPOVERLAPPED lpOverlapped)
+{
+	ILibProcessPipe_PipeObject *j = (ILibProcessPipe_PipeObject*)((void**)ILibMemory_GetExtraMemory(lpOverlapped, sizeof(OVERLAPPED)))[0];
+	if (!ILibMemory_CanaryOK(j)) { return; }
+	
+	ILibProcessPipe_Pipe_ReadExHandler callback = (ILibProcessPipe_Pipe_ReadExHandler)j->user2;
+	if (callback != NULL) { callback(j, j->user1, dwErrorCode, j->buffer, dwNumberOfBytesTransfered); }
+}
+void __stdcall ILibProcessPipe_Pipe_Write_CompletionRoutine(DWORD dwErrorCode, DWORD dwNumberOfBytesTransfered, LPOVERLAPPED lpOverlapped)
+{
+	ILibProcessPipe_PipeObject *j = (ILibProcessPipe_PipeObject*)((void**)ILibMemory_GetExtraMemory(lpOverlapped, sizeof(OVERLAPPED)))[0];
+	if (!ILibMemory_CanaryOK(j)) { return; }
+
+	if (j->user4 != NULL)
+	{
+		((ILibProcessPipe_Pipe_WriteExHandler)j->user4)(j, j->user3, dwErrorCode, dwNumberOfBytesTransfered);
+	}
+}
+void ILibProcessPipe_Pipe_ReadEx(ILibProcessPipe_Pipe targetPipe, char *buffer, int bufferLength, void *user, ILibProcessPipe_Pipe_ReadExHandler OnReadHandler)
+{
+	ILibProcessPipe_PipeObject *j = (ILibProcessPipe_PipeObject*)targetPipe;
+	j->buffer = buffer;
+	j->bufferSize = bufferLength;
+	j->user1 = user;
+	j->user2 = OnReadHandler;
+	ReadFileEx(j->mPipe_ReadEnd, j->buffer, j->bufferSize, j->mOverlapped, ILibProcessPipe_Pipe_Read_CompletionRoutine);
+}
+void ILibProcessPipe_Pipe_WriteEx(ILibProcessPipe_Pipe targetPipe, char *buffer, int bufferLength, void *user, ILibProcessPipe_Pipe_WriteExHandler OnWriteHandler)
+{
+	ILibProcessPipe_PipeObject *j = (ILibProcessPipe_PipeObject*)targetPipe;
+	if (j->mwOverlapped == NULL)
+	{
+		j->mwOverlapped = (OVERLAPPED*)ILibMemory_Allocate(sizeof(OVERLAPPED), 0, NULL, NULL);
+	}
+	j->user3 = user;
+	j->user4 = OnWriteHandler;
+	WriteFileEx(j->mPipe_WriteEnd, buffer, bufferLength, j->mwOverlapped, ILibProcessPipe_Pipe_Write_CompletionRoutine);
+}
 DWORD ILibProcessPipe_Process_GetPID(ILibProcessPipe_Process p) { return(p != NULL ? (DWORD)((ILibProcessPipe_Process_Object*)p)->PID : 0); }
 #else
 pid_t ILibProcessPipe_Process_GetPID(ILibProcessPipe_Process p) { return(p != NULL ? (pid_t)((ILibProcessPipe_Process_Object*)p)->PID : 0); }
--- a/microstack/ILibProcessPipe.h
+++ b/microstack/ILibProcessPipe.h
@@ -44,6 +44,8 @@ typedef enum ILibProcessPipe_SpawnTypes
 }ILibProcessPipe_SpawnTypes;

 #ifdef WIN32
+typedef void(*ILibProcessPipe_Pipe_ReadExHandler)(ILibProcessPipe_Pipe sender, void *user, DWORD errorCode, char *buffer, int bufferLen);
+typedef void(*ILibProcessPipe_Pipe_WriteExHandler)(ILibProcessPipe_Pipe sender, void *user, DWORD errorCode, int bytesWritten);
 typedef enum ILibProcessPipe_Pipe_ReaderHandleType
 {
 	ILibProcessPipe_Pipe_ReaderHandleType_NotOverLapped = 0,	//!< Spawn a I/O processing thread
@@ -55,6 +57,8 @@ HANDLE ILibProcessPipe_Manager_GetWorkerThread(ILibProcessPipe_Manager mgr);
 ILibTransport_DoneState ILibProcessPipe_Pipe_Write(ILibProcessPipe_Pipe writePipe, char* buffer, int bufferLen, ILibTransport_MemoryOwnership ownership);
 void ILibProcessPipe_Pipe_AddPipeReadHandler(ILibProcessPipe_Pipe targetPipe, int bufferSize, ILibProcessPipe_Pipe_ReadHandler OnReadHandler);
 #ifdef WIN32
+	void ILibProcessPipe_Pipe_ReadEx(ILibProcessPipe_Pipe targetPipe, char *buffer, int bufferLength, void *user, ILibProcessPipe_Pipe_ReadExHandler OnReadHandler);
+	void ILibProcessPipe_Pipe_WriteEx(ILibProcessPipe_Pipe targetPipe, char *buffer, int bufferLength, void *user, ILibProcessPipe_Pipe_WriteExHandler OnWriteHandler);
 	ILibProcessPipe_Pipe ILibProcessPipe_Pipe_CreateFromExistingWithExtraMemory(ILibProcessPipe_Manager manager, HANDLE existingPipe, ILibProcessPipe_Pipe_ReaderHandleType handleType, int extraMemorySize);
 	#define ILibProcessPipe_Pipe_CreateFromExisting(PipeManager, ExistingPipe, HandleType) ILibProcessPipe_Pipe_CreateFromExistingWithExtraMemory(PipeManager, ExistingPipe, HandleType, 0)
 #else
@@ -100,6 +104,9 @@ typedef BOOL(*ILibProcessPipe_WaitHandle_Handler)(HANDLE event, ILibWaitHandle_E
 void ILibProcessPipe_WaitHandle_Add_WithNonZeroTimeout(ILibProcessPipe_Manager mgr, HANDLE event, int milliseconds, void *user, ILibProcessPipe_WaitHandle_Handler callback);
 #define ILibProcessPipe_WaitHandle_Add(processPipeManager, eventHandle, user, callback) ILibProcessPipe_WaitHandle_Add_WithNonZeroTimeout(processPipeManager, eventHandle, 0, user, callback)
 void ILibProcessPipe_WaitHandle_Remove(ILibProcessPipe_Manager mgr, HANDLE event);
+void ILibProcessPipe_WaitHandle_Add2_WithNonZeroTimeout(ILibProcessPipe_Manager mgr, HANDLE event, int milliseconds, void *user, ILibProcessPipe_WaitHandle_Handler callback);
+#define ILibProcessPipe_WaitHandle_Add2(processPipeManager, eventHandle, user, callback) ILibProcessPipe_WaitHandle_Add2_WithNonZeroTimeout(processPipeManager, eventHandle, 0, user, callback)
+
 #endif
 #define ILibTransports_ProcessPipe 0x60
 #endif