MeshCentral/MeshAgent
1
0
Fork 0
mirror of https://github.com/Ylianst/MeshAgent synced 2025-12-30 15:13:57 +00:00
Code Issues Releases Wiki Activity

Updated Windows Chain logic, to use WSAEventSelect and WaitForMultipleObjectsEx

Browse Source
This commit is contained in:
Bryan Roe
2020-05-04 17:25:55 -07:00
parent 11da1118b4
commit 1bd4e3d07c
5 changed files with 207 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
microscript/ILibDuktape_net.c
View File

@@ -1079,9 +1079,10 @@ duk_ret_t ILibDuktape_net_server_IPC_ConnectSink_Finalizer(duk_context *ctx)
}
return(0);
}
BOOL ILibDuktape_net_server_IPC_ConnectSink(HANDLE event, ILibWaitHandle_ErrorStatus status, void* user)
//BOOL ILibDuktape_net_server_IPC_ConnectSink(HANDLE event, ILibWaitHandle_ErrorStatus status, void* user)
BOOL ILibDuktape_net_server_IPC_ConnectSink(void *chain, HANDLE event, ILibWaitHandle_ErrorStatus status, void* user)
{
if (ILibMemory_CanaryOK(user))
if (ILibMemory_CanaryOK(user) && status == ILibWaitHandle_ErrorStatus_NONE)
{
ILibDuktape_net_WindowsIPC *winIPC = (ILibDuktape_net_WindowsIPC*)user;
ILibDuktape_EventEmitter_SetupEmit(winIPC->ctx, winIPC->mServer, "connection"); // [emit][this][connection]
@@ -1253,8 +1254,10 @@ duk_ret_t ILibDuktape_net_server_listen(duk_context *ctx)
duk_del_prop_string(ctx, -1, ILibDuktape_net_WindowsIPC_Buffer);
return(ILibDuktape_Error(ctx, "Error Creating Named Pipe: %s", ipc));
}
//printf("ConnectNamedPipe(%s)\n", ipc);
ConnectNamedPipe(winIPC->mPipeHandle, &winIPC->overlapped);
ILibProcessPipe_WaitHandle_Add2(winIPC->manager, winIPC->overlapped.hEvent, winIPC, ILibDuktape_net_server_IPC_ConnectSink);
//ILibProcessPipe_WaitHandle_Add2(winIPC->manager, winIPC->overlapped.hEvent, winIPC, ILibDuktape_net_server_IPC_ConnectSink);
ILibChain_AddWaitHandle(duk_ctx_chain(ctx), winIPC->overlapped.hEvent, -1, ILibDuktape_net_server_IPC_ConnectSink, winIPC);
if (pIPC_SA != NULL) { LocalFree(IPC_ACL); }
return(1);

215
microstack/ILibParsers.c
View File

@@ -180,6 +180,7 @@ int ILibWhichPowerOfTwo(int number)
}
#ifdef WIN32
#define ILibChain_HandleInfoIndex(i) (FD_SETSIZE+i)
WCHAR ILibWideScratchPad[4096];
char ILibUTF8ScratchPad[4096];
char *ILibWideToUTF8Ex(WCHAR* wstr, int len, char *buffer, int bufferLen)
@@ -948,7 +949,8 @@ typedef struct ILibBaseChain
pthread_t ChainThreadID;
#endif
#if defined(WIN32)
SOCKET TerminateSock;
int UnblockFlag;
void* auxSelectHandles;
#else
int TerminatePipe[2];
#endif
@@ -973,6 +975,14 @@ typedef struct ILibBaseChain
void *node;
}ILibBaseChain;
typedef struct ILibChain_WaitHandleInfo
{
void *node;
ILibChain_WaitHandleHandler handler;
void *user;
struct timeval expiration;
}ILibChain_WaitHandleInfo;
void* ILibMemory_AllocateA_Init(void *buffer)
{
((void**)buffer)[0] = (char*)buffer + sizeof(void*);
@@ -1834,7 +1844,6 @@ void *ILibCreateChainEx(int extraMemorySize)
#if defined(WIN32) || defined(_WIN32_WCE)
RetVal->TerminateFlag = 1;
RetVal->TerminateSock = WSASocketW(AF_INET, SOCK_DGRAM, IPPROTO_UDP, NULL, 0, WSA_FLAG_NO_HANDLE_INHERIT);
RetVal->ChainProcessHandle = GetCurrentProcess();
if (!SymInitialize(RetVal->ChainProcessHandle, NULL, TRUE)) { RetVal->ChainProcessHandle = NULL; }
#endif
@@ -1902,12 +1911,8 @@ void* ILibGetBaseTimer(void *chain)
void __stdcall ILibForceUnBlockChain_APC(ULONG_PTR obj)
{
struct ILibBaseChain *c = (struct ILibBaseChain*)obj;
c->selectTimeout = 0;
if (c->TerminateSock != ~0)
{
closesocket(c->TerminateSock);
}
c->TerminateSock = WSASocketW(AF_INET, SOCK_DGRAM, IPPROTO_UDP, NULL, 0, WSA_FLAG_NO_HANDLE_INHERIT);
//c->selectTimeout = 0;
c->UnblockFlag = 1;
}
#endif
/*! \fn ILibForceUnBlockChain(void *Chain)
@@ -2061,12 +2066,7 @@ ILibExportMethod void ILibChain_Continue(void *Chain, ILibChain_Link **modules,
tv.tv_usec = 1000 * (chain->selectTimeout % 1000);
#if defined(WIN32) || defined(_WIN32_WCE)
//
// Add the fake socket, for ILibForceUnBlockChain
//
FD_SET(chain->TerminateSock, &errorset);
#else
#if !defined(WIN32)
//
// Put the Read end of the Pipe in the FDSET, for ILibForceUnBlockChain
//
@@ -2762,9 +2762,7 @@ char *ILibChain_GetMetaDataFromDescriptorSet(void *chain, fd_set *inr, fd_set *i
if (len > 0) { ret = ILibScratchPad; }
if (ret == NULL)
{
#if defined(WIN32)
if (FD_ISSET(((ILibBaseChain*)chain)->TerminateSock, ine) || FD_ISSET(((ILibBaseChain*)chain)->TerminateSock, inr)) { ret = "ILibForceUnblockChain"; }
#else
#if !defined(WIN32)
if (FD_ISSET(((ILibBaseChain*)chain)->TerminatePipe[0], inr)) { ret = "ILibForceUnblockChain"; }
#endif
if (ret == NULL)
@@ -2806,6 +2804,25 @@ void *ILibChain_GetObjectForDescriptor(void *chain, int fd)
return(ret);
}
#ifdef WIN32
void ILibChain_AddWaitHandle(void *chain, HANDLE h, int msTIMEOUT, ILibChain_WaitHandleHandler handler, void *user)
{
void *node = ILibLinkedList_AddTail(((ILibBaseChain*)chain)->auxSelectHandles, h);
ILibChain_WaitHandleInfo *info = (ILibChain_WaitHandleInfo*)ILibMemory_Extra(node);
info->handler = handler;
info->user = user;
info->node = node;
if (msTIMEOUT != INFINITE && msTIMEOUT >= 0)
{
ILibGetTimeOfDay(&(info->expiration));
info->expiration.tv_sec += (long)(msTIMEOUT / 1000);
info->expiration.tv_usec += ((msTIMEOUT % 1000) * 1000);
}
ILibForceUnBlockChain(chain);
}
#endif
/*! \fn ILibStartChain(void *Chain)
\brief Starts a Chain
\par
@@ -2824,7 +2841,17 @@ ILibExportMethod void ILibStartChain(void *Chain)
fd_set errorset;
fd_set writeset;
struct timeval tv;
#ifdef WIN32
void *node;
DWORD waitTimeout;
HANDLE selectHandles[FD_SETSIZE];
HANDLE selectEvents[FD_SETSIZE*2];
memset(selectHandles, 0, sizeof(selectHandles));
memset(selectEvents, 0, sizeof(selectEvents));
struct timeval currentTime;
struct timeval expirationTime;
#endif
struct timeval tv;
int slct;
int vX;
@@ -2835,6 +2862,7 @@ ILibExportMethod void ILibStartChain(void *Chain)
chain->ChainThreadID = GetCurrentThreadId();
chain->ChainProcessHandle = GetCurrentProcess();
chain->MicrostackThreadHandle = OpenThread(THREAD_ALL_ACCESS, FALSE, chain->ChainThreadID);
chain->auxSelectHandles = ILibLinkedList_CreateEx(sizeof(ILibChain_WaitHandleInfo));
#else
chain->ChainThreadID = pthread_self();
#endif
@@ -2921,12 +2949,7 @@ ILibExportMethod void ILibStartChain(void *Chain)
tv.tv_usec = 1000 * (chain->selectTimeout % 1000);
#if defined(WIN32) || defined(_WIN32_WCE)
//
// Add the fake socket, for ILibForceUnBlockChain
//
FD_SET(chain->TerminateSock, &errorset);
#else
#if !defined(WIN32)
//
// Put the Read end of the Pipe in the FDSET, for ILibForceUnBlockChain
//
@@ -2952,14 +2975,139 @@ ILibExportMethod void ILibStartChain(void *Chain)
//
chain->PreSelectCount++;
#ifdef WIN32
if (readset.fd_count == 0 && writeset.fd_count == 0)
if (readset.fd_count == 0 && writeset.fd_count == 0 && ILibLinkedList_GetNode_Head(chain->auxSelectHandles)==NULL)
{
SleepEx((DWORD)chain->selectTimeout, TRUE); // If there is no pending IO, we must force the thread into an alertable wait state, so ILibForceUnblockChain can function.
slct = -1;
}
else
{
int i;
int x = 0;
long flags;
for (i = 0; i < (int)readset.fd_count; ++i)
{
selectHandles[x++] = (HANDLE)readset.fd_array[i];
selectEvents[ILibChain_HandleInfoIndex(x - 1)] = NULL;
}
for (i = 0; i < (int)writeset.fd_count; ++i)
{
if (!FD_ISSET(writeset.fd_array[i], &readset))
{
selectHandles[x++] = (HANDLE)writeset.fd_array[i];
selectEvents[ILibChain_HandleInfoIndex(x - 1)] = NULL;
}
}
for (i = 0; i < (int)errorset.fd_count; ++i)
{
if (!FD_ISSET(errorset.fd_array[i], &readset) && !FD_ISSET(errorset.fd_array[i], &writeset))
{
selectHandles[x++] = (HANDLE)errorset.fd_array[i];
selectEvents[ILibChain_HandleInfoIndex(x - 1)] = NULL;
}
}
for (i = 0; i < x; ++i)
{
if (selectEvents[i] == NULL || selectEvents[ILibChain_HandleInfoIndex(i)] != NULL)
{
selectEvents[i] = WSACreateEvent();
}
else
{
WSAResetEvent(selectEvents[i]);
}
flags = 0;
if (FD_ISSET(selectHandles[i], &readset)) { flags |= (FD_READ | FD_ACCEPT); }
if (FD_ISSET(selectHandles[i], &writeset)) { flags |= (FD_WRITE | FD_CONNECT); }
if (FD_ISSET(selectHandles[i], &errorset)) { flags |= FD_CLOSE; }
WSAEventSelect((SOCKET)selectHandles[i], selectEvents[i], flags);
}
ILibGetTimeOfDay(&currentTime);
memcpy_s(&expirationTime, sizeof(struct timeval), &currentTime, sizeof(struct timeval));
expirationTime.tv_sec += tv.tv_sec;
expirationTime.tv_usec += tv.tv_usec;
node = ILibLinkedList_GetNode_Head(chain->auxSelectHandles);
while (node != NULL)
{
i = x++;
if (selectEvents[i] != NULL && selectEvents[ILibChain_HandleInfoIndex(i)] == NULL) { WSACloseEvent(selectEvents[i]); }
selectEvents[i] = (HANDLE)ILibLinkedList_GetDataFromNode(node);
selectEvents[ILibChain_HandleInfoIndex(i)] = (HANDLE)ILibMemory_Extra(node);
if (((ILibChain_WaitHandleInfo*)ILibMemory_Extra(node))->expiration.tv_sec != 0 || ((ILibChain_WaitHandleInfo*)ILibMemory_Extra(node))->expiration.tv_usec != 0)
{
// Timeout was specified
if (tv2LTtv1(&expirationTime, &(((ILibChain_WaitHandleInfo*)ILibMemory_Extra(node))->expiration)))
{
expirationTime.tv_sec = ((ILibChain_WaitHandleInfo*)ILibMemory_Extra(node))->expiration.tv_sec;
expirationTime.tv_usec = ((ILibChain_WaitHandleInfo*)ILibMemory_Extra(node))->expiration.tv_usec;
// If the expiration happens in the past, we need to set the timeout to zero
if (tv2LTtv1(&expirationTime, &currentTime)) { expirationTime.tv_sec = currentTime.tv_sec; expirationTime.tv_usec = currentTime.tv_usec; }
}
}
node = ILibLinkedList_GetNextNode(node);
}
expirationTime.tv_sec -= currentTime.tv_sec;
expirationTime.tv_usec -= currentTime.tv_usec;
waitTimeout = (DWORD)((expirationTime.tv_sec * 1000) + (expirationTime.tv_usec / 0.001));
while ((slct = WaitForMultipleObjectsEx(x, selectEvents, FALSE, (DWORD)((tv.tv_sec * 1000) + (tv.tv_usec / 0.001)), TRUE)) == WAIT_IO_COMPLETION && chain->UnblockFlag == 0) {}
ILibGetTimeOfDay(&currentTime);
if (slct != WAIT_IO_COMPLETION && (slct - WAIT_OBJECT_0 >= 0) && (slct - WAIT_OBJECT_0 < x))
{
if (selectEvents[ILibChain_HandleInfoIndex(slct)] != NULL)
{
ILibChain_WaitHandleInfo *info = (ILibChain_WaitHandleInfo*)selectEvents[ILibChain_HandleInfoIndex(slct)];
selectEvents[ILibChain_HandleInfoIndex(slct)] = NULL;
if (info->handler != NULL)
{
if (info->handler(chain, selectEvents[slct], ILibWaitHandle_ErrorStatus_NONE, info->user) == FALSE)
{
// FALSE means to remove tha HANDLE
ILibLinkedList_Remove(info->node);
}
}
}
}
if (slct == WAIT_TIMEOUT)
{
for (i = 0; i < x; ++i)
{
if (selectEvents[ILibChain_HandleInfoIndex(i)] != NULL && tvnonzero(&(((ILibChain_WaitHandleInfo*)selectEvents[ILibChain_HandleInfoIndex(i)])->expiration)))
{
if (tv2LTEtv1(&currentTime, &(((ILibChain_WaitHandleInfo*)selectEvents[ILibChain_HandleInfoIndex(i)])->expiration)))
{
// TIMEOUT occured
if (((ILibChain_WaitHandleInfo*)selectEvents[ILibChain_HandleInfoIndex(i)])->handler != NULL)
{
((ILibChain_WaitHandleInfo*)selectEvents[ILibChain_HandleInfoIndex(i)])->handler(chain, selectEvents[i], ILibWaitHandle_ErrorStatus_TIMEOUT, ((ILibChain_WaitHandleInfo*)selectEvents[ILibChain_HandleInfoIndex(i)])->user);
}
ILibLinkedList_Remove(((ILibChain_WaitHandleInfo*)selectEvents[ILibChain_HandleInfoIndex(i)])->node);
}
}
}
}
if (slct == WAIT_FAILED)
{
// One of the handles is invalid... Kick it out
for (i = 0; i < x; ++i)
{
if (selectEvents[ILibChain_HandleInfoIndex(i)] != NULL)
{
if (WaitForSingleObject(selectEvents[i], 0) == WAIT_FAILED)
{
if (((ILibChain_WaitHandleInfo*)selectEvents[ILibChain_HandleInfoIndex(i)])->handler != NULL)
{
((ILibChain_WaitHandleInfo*)selectEvents[ILibChain_HandleInfoIndex(i)])->handler(chain, selectEvents[i], ILibWaitHandle_ErrorStatus_INVALID_HANDLE, ((ILibChain_WaitHandleInfo*)selectEvents[ILibChain_HandleInfoIndex(i)])->user);
}
ILibLinkedList_Remove(((ILibChain_WaitHandleInfo*)selectEvents[ILibChain_HandleInfoIndex(i)])->node);
}
}
}
}
tv.tv_sec = 0; tv.tv_usec = 0;
slct = select(FD_SETSIZE, &readset, &writeset, &errorset, &tv);
chain->UnblockFlag = 0;
}
#else
slct = select(FD_SETSIZE, &readset, &writeset, &errorset, &tv);
@@ -3083,6 +3231,16 @@ ILibExportMethod void ILibStartChain(void *Chain)
chain->node = ILibLinkedList_Remove(chain->node);
}
for (vX = 0; vX < FD_SETSIZE; ++vX)
{
if (selectHandles[vX] != NULL && selectHandles[ILibChain_HandleInfoIndex(vX)] == NULL)
{
WSACloseEvent(selectHandles[vX]);
}
}
ILibLinkedList_Destroy(chain->auxSelectHandles);
if (gILibChain == Chain) { gILibChain = NULL; } // Reset the global instance if it was set
//
@@ -3129,13 +3287,6 @@ ILibExportMethod void ILibStartChain(void *Chain)
((ILibBaseChain*)Chain)->TerminatePipe[0] = 0;
((ILibBaseChain*)Chain)->TerminatePipe[1] = 0;
#endif
#if defined(WIN32)
if (((ILibBaseChain*)Chain)->TerminateSock != ~0)
{
closesocket(((ILibBaseChain*)Chain)->TerminateSock);
((ILibBaseChain*)Chain)->TerminateSock = (SOCKET)~0;
}
#endif
#ifdef WIN32
WSACleanup();

18
microstack/ILibParsers.h
View File

@@ -347,6 +347,17 @@ int ILibIsRunningOnChainThread(void* chain);
typedef void*(*ILibChain_Link_GetUserMemory)(void *ChainLinkObject, int *len);
typedef void* ILibChain_EventHookToken;
typedef void(*ILibChain_EventHookHandler)(void *hookedObject, ILibChain_EventHookToken token);
#ifdef WIN32
typedef enum ILibWaitHandle_ErrorStatus
{
ILibWaitHandle_ErrorStatus_NONE = 0,
ILibWaitHandle_ErrorStatus_INVALID_HANDLE = 1,
ILibWaitHandle_ErrorStatus_TIMEOUT = 2,
ILibWaitHandle_ErrorStatus_REMOVED = 3,
ILibWaitHandle_ErrorStatus_MANAGER_EXITING = 4
}ILibWaitHandle_ErrorStatus;
typedef BOOL(*ILibChain_WaitHandleHandler)(void *chain, HANDLE h, ILibWaitHandle_ErrorStatus, void* user);
#endif
typedef struct ILibChain_Link
{
@@ -961,6 +972,13 @@ int ILibIsRunningOnChainThread(void* chain);
void ILibChain_DisableWatchDog(void *chain);
void *ILibChain_GetObjectForDescriptor(void *chain, int fd);
char *ILibChain_GetMetaDataFromDescriptorSet(void *chain, fd_set *inr, fd_set *inw, fd_set *ine);
#ifdef WIN32
void ILibChain_AddWaitHandle(void *chain, HANDLE h, int msTIMEOUT, ILibChain_WaitHandleHandler handler, void *user);
#define tv2LTtv1(ptv1, ptv2) ((ptv2)->tv_sec < (ptv1)->tv_sec || ((ptv2)->tv_sec == (ptv1)->tv_sec && (ptv2)->tv_usec < (ptv1)->tv_usec))
#define tv2LTEtv1(ptv1, ptv2) (tv2LTtv1(ptv2,ptv1) || ((ptv2)->tv_sec == (ptv1)->tv_sec && (ptv2)->tv_usec <= (ptv1)->tv_usec))
#define tvnonzero(ptv) ((ptv)->tv_sec != 0 || (ptv)->tv_usec != 0)
#endif
ILibExportMethod void ILibStartChain(void *chain);
ILibExportMethod void ILibStopChain(void *chain);
ILibExportMethod void ILibChain_Continue(void *chain, ILibChain_Link **modules, int moduleCount, int maxTimeout);

33
microstack/ILibProcessPipe.c
View File

@@ -259,39 +259,6 @@ void ILibProcessPipe_WaitHandle_Add_WithNonZeroTimeout(ILibProcessPipe_Manager m
ILibProcessPipe_WaitHandle_AddEx(mgr, waitHandle);
}
void ILibProcessPipe_WaitHandle_Add2_chainsink(void *chain, void *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))
{
if (status == ILibWaitHandle_ErrorStatus_REMOVED || status == ILibWaitHandle_ErrorStatus_MANAGER_EXITING)
{
ILibMemory_Free(user);
}
else
{
ILibChain_RunOnMicrostackThread(((ILibProcessPipe_WaitHandle_APC*)user)->chain, ILibProcessPipe_WaitHandle_Add2_chainsink, 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;
apcState->chain = ((ILibProcessPipe_Manager_Object*)mgr)->ChainLink.ParentChain;
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;

13
microstack/ILibProcessPipe.h
View File

@@ -102,15 +102,6 @@ int ILibProcessPipe_Process_GetPTY(ILibProcessPipe_Process p);
#ifdef WIN32
typedef enum ILibWaitHandle_ErrorStatus
{
ILibWaitHandle_ErrorStatus_NONE = 0,
ILibWaitHandle_ErrorStatus_INVALID_HANDLE = 1,
ILibWaitHandle_ErrorStatus_TIMEOUT = 2,
ILibWaitHandle_ErrorStatus_REMOVED = 3,
ILibWaitHandle_ErrorStatus_MANAGER_EXITING = 4
}ILibWaitHandle_ErrorStatus;
typedef BOOL(*ILibProcessPipe_WaitHandle_Handler)(HANDLE event, ILibWaitHandle_ErrorStatus status, void* user);
void ILibProcessPipe_WaitHandle_Remove(ILibProcessPipe_Manager mgr, HANDLE event);
@@ -118,10 +109,6 @@ void ILibProcessPipe_WaitHandle_Remove(ILibProcessPipe_Manager mgr, HANDLE event
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)
// These methods will context switch to the chain thread when dispatching
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