Skip to content

P
Postgres FD Implementation
Commit 70d58115 authored by Bruce Momjian's avatar Bruce Momjian
Browse files
Options

Pgindent win32 signal code.

parent 50491963
Hide whitespace changes
Inline Side-by-side
Showing with 214 additions and 173 deletions
src/backend/libpq/pqsignal.c
View file @ 70d58115
...@@ -9,7 +9,7 @@ ...@@ -9,7 +9,7 @@
* *
* *
* IDENTIFICATION * IDENTIFICATION
* $PostgreSQL: pgsql/src/backend/libpq/pqsignal.c,v 1.29 2004/01/27 00:45:26 momjian Exp $ * $PostgreSQL: pgsql/src/backend/libpq/pqsignal.c,v 1.30 2004/01/27 00:46:58 momjian Exp $
* *
* NOTES * NOTES
* This shouldn't be in libpq, but the monitor and some other * This shouldn't be in libpq, but the monitor and some other
...@@ -169,141 +169,171 @@ pqsignal(int signo, pqsigfunc func) ...@@ -169,141 +169,171 @@ pqsignal(int signo, pqsigfunc func)
/* pg_signal_crit_sec is used to protect only pg_signal_queue. That is the only /* pg_signal_crit_sec is used to protect only pg_signal_queue. That is the only
* variable that can be accessed from the signal sending threads! */ * variable that can be accessed from the signal sending threads! */
static CRITICAL_SECTION pg_signal_crit_sec; static CRITICAL_SECTION pg_signal_crit_sec;
static int pg_signal_queue; static int pg_signal_queue;
#define PG_SIGNAL_COUNT 32 #define PG_SIGNAL_COUNT 32
static pqsigfunc pg_signal_array[PG_SIGNAL_COUNT]; static pqsigfunc pg_signal_array[PG_SIGNAL_COUNT];
static pqsigfunc pg_signal_defaults[PG_SIGNAL_COUNT]; static pqsigfunc pg_signal_defaults[PG_SIGNAL_COUNT];
static int pg_signal_mask; static int pg_signal_mask;
HANDLE pgwin32_main_thread_handle; HANDLE pgwin32_main_thread_handle;
/* Signal handling thread function */ /* Signal handling thread function */
static DWORD WINAPI pg_signal_thread(LPVOID param); static DWORD WINAPI pg_signal_thread(LPVOID param);
/* Initialization */ /* Initialization */
void pgwin32_signal_initialize(void) { void
int i; pgwin32_signal_initialize(void)
HANDLE signal_thread_handle; {
InitializeCriticalSection(&pg_signal_crit_sec); int i;
HANDLE signal_thread_handle;
for (i = 0; i < PG_SIGNAL_COUNT; i++) {
pg_signal_array[i] = SIG_DFL; InitializeCriticalSection(&pg_signal_crit_sec);
pg_signal_defaults[i] = SIG_IGN;
} for (i = 0; i < PG_SIGNAL_COUNT; i++)
pg_signal_mask = 0; {
pg_signal_queue = 0; pg_signal_array[i] = SIG_DFL;
pg_signal_defaults[i] = SIG_IGN;
/* Get handle to main thread so we can post calls to it later */ }
if (!DuplicateHandle(GetCurrentProcess(),GetCurrentThread(), pg_signal_mask = 0;
GetCurrentProcess(),&pgwin32_main_thread_handle, pg_signal_queue = 0;
0,FALSE,DUPLICATE_SAME_ACCESS)) {
fprintf(stderr,gettext("Failed to get main thread handle!\n")); /* Get handle to main thread so we can post calls to it later */
exit(1); if (!DuplicateHandle(GetCurrentProcess(), GetCurrentThread(),
} GetCurrentProcess(), &pgwin32_main_thread_handle,
0, FALSE, DUPLICATE_SAME_ACCESS))
/* Create thread for handling signals */ {
signal_thread_handle = CreateThread(NULL,0,pg_signal_thread,NULL,0,NULL); fprintf(stderr, gettext("Failed to get main thread handle!\n"));
if (signal_thread_handle == NULL) { exit(1);
fprintf(stderr,gettext("Failed to create signal handler thread!\n")); }
exit(1);
} /* Create thread for handling signals */
signal_thread_handle = CreateThread(NULL, 0, pg_signal_thread, NULL, 0, NULL);
if (signal_thread_handle == NULL)
{
fprintf(stderr, gettext("Failed to create signal handler thread!\n"));
exit(1);
}
} }
/* Dispatch all signals currently queued and not blocked /* Dispatch all signals currently queued and not blocked
* Blocked signals are ignored, and will be fired at the time of * Blocked signals are ignored, and will be fired at the time of
* the sigsetmask() call. */ * the sigsetmask() call. */
static void dispatch_queued_signals(void) { static void
int i; dispatch_queued_signals(void)
{
EnterCriticalSection(&pg_signal_crit_sec); int i;
while (pg_signal_queue & ~pg_signal_mask) {
/* One or more unblocked signals queued for execution */ EnterCriticalSection(&pg_signal_crit_sec);
while (pg_signal_queue & ~pg_signal_mask)
int exec_mask = pg_signal_queue & ~pg_signal_mask; {
/* One or more unblocked signals queued for execution */
for (i = 0; i < PG_SIGNAL_COUNT; i++) {
if (exec_mask & sigmask(i)) { int exec_mask = pg_signal_queue & ~pg_signal_mask;
/* Execute this signal */
pqsigfunc sig = pg_signal_array[i]; for (i = 0; i < PG_SIGNAL_COUNT; i++)
if (sig == SIG_DFL) {
sig = pg_signal_defaults[i]; if (exec_mask & sigmask(i))
pg_signal_queue &= ~sigmask(i); {
if (sig != SIG_ERR && sig != SIG_IGN && sig != SIG_DFL) { /* Execute this signal */
LeaveCriticalSection(&pg_signal_crit_sec); pqsigfunc sig = pg_signal_array[i];
sig(i);
EnterCriticalSection(&pg_signal_crit_sec); if (sig == SIG_DFL)
break; /* Restart outer loop, in case signal mask or queue sig = pg_signal_defaults[i];
has been modified inside signal handler */ pg_signal_queue &= ~sigmask(i);
if (sig != SIG_ERR && sig != SIG_IGN && sig != SIG_DFL)
{
LeaveCriticalSection(&pg_signal_crit_sec);
sig(i);
EnterCriticalSection(&pg_signal_crit_sec);
break; /* Restart outer loop, in case signal mask
* or queue has been modified inside
* signal handler */
}
}
}
} }
} LeaveCriticalSection(&pg_signal_crit_sec);
}
}
LeaveCriticalSection(&pg_signal_crit_sec);
} }
/* signal masking. Only called on main thread, no sync required */ /* signal masking. Only called on main thread, no sync required */
int pqsigsetmask(int mask) { int
int prevmask; pqsigsetmask(int mask)
prevmask = pg_signal_mask; {
pg_signal_mask = mask; int prevmask;
/* Dispatch any signals queued up right away, in case we have prevmask = pg_signal_mask;
unblocked one or more signals previously queued */ pg_signal_mask = mask;
dispatch_queued_signals();
/*
return prevmask; * Dispatch any signals queued up right away, in case we have
* unblocked one or more signals previously queued
*/
dispatch_queued_signals();
return prevmask;
} }
/* signal manipulation. Only called on main thread, no sync required */ /* signal manipulation. Only called on main thread, no sync required */
pqsigfunc pqsignal(int signum, pqsigfunc handler) { pqsigfunc
pqsigfunc prevfunc; pqsignal(int signum, pqsigfunc handler)
if (signum >= PG_SIGNAL_COUNT || signum < 0) {
return SIG_ERR; pqsigfunc prevfunc;
prevfunc = pg_signal_array[signum];
pg_signal_array[signum] = handler; if (signum >= PG_SIGNAL_COUNT || signum < 0)
return prevfunc; return SIG_ERR;
prevfunc = pg_signal_array[signum];
pg_signal_array[signum] = handler;
return prevfunc;
} }
/* signal sending */ /* signal sending */
int pqkill(int pid, int sig) { int
char pipename[128]; pqkill(int pid, int sig)
BYTE sigData = sig; {
BYTE sigRet = 0; char pipename[128];
DWORD bytes; BYTE sigData = sig;
BYTE sigRet = 0;
if (sig >= PG_SIGNAL_COUNT || sig <= 0) { DWORD bytes;
errno = EINVAL;
return -1; if (sig >= PG_SIGNAL_COUNT || sig <= 0)
} {
if (pid <= 0) { errno = EINVAL;
/* No support for process groups */ return -1;
errno = EINVAL; }
return -1; if (pid <= 0)
} {
wsprintf(pipename,"\\\\.\\pipe\\pgsignal_%i",pid); /* No support for process groups */
if (!CallNamedPipe(pipename,&sigData,1,&sigRet,1,&bytes,1000)) { errno = EINVAL;
if (GetLastError() == ERROR_FILE_NOT_FOUND) return -1;
errno = ESRCH; }
else if (GetLastError() == ERROR_ACCESS_DENIED) wsprintf(pipename, "\\\\.\\pipe\\pgsignal_%i", pid);
errno = EPERM; if (!CallNamedPipe(pipename, &sigData, 1, &sigRet, 1, &bytes, 1000))
else {
errno = EINVAL; if (GetLastError() == ERROR_FILE_NOT_FOUND)
return -1; errno = ESRCH;
} else if (GetLastError() == ERROR_ACCESS_DENIED)
if (bytes != 1 || sigRet != sig) { errno = EPERM;
errno = ESRCH; else
return -1; errno = EINVAL;
} return -1;
}
return 0; if (bytes != 1 || sigRet != sig)
{
errno = ESRCH;
return -1;
}
return 0;
} }
/* APC callback scheduled on main thread when signals are fired */ /* APC callback scheduled on main thread when signals are fired */
static void CALLBACK pg_signal_apc(ULONG_PTR param) { static void CALLBACK
dispatch_queued_signals(); pg_signal_apc(ULONG_PTR param)
{
dispatch_queued_signals();
} }
/* /*
...@@ -314,78 +344,89 @@ static void CALLBACK pg_signal_apc(ULONG_PTR param) { ...@@ -314,78 +344,89 @@ static void CALLBACK pg_signal_apc(ULONG_PTR param) {
*/ */
static void pg_queue_signal(int signum) { static void
if (signum >= PG_SIGNAL_COUNT || signum < 0) pg_queue_signal(int signum)
return; {
if (signum >= PG_SIGNAL_COUNT || signum < 0)
EnterCriticalSection(&pg_signal_crit_sec); return;
pg_signal_queue |= sigmask(signum);
LeaveCriticalSection(&pg_signal_crit_sec); EnterCriticalSection(&pg_signal_crit_sec);
pg_signal_queue |= sigmask(signum);
QueueUserAPC(pg_signal_apc,pgwin32_main_thread_handle,(ULONG_PTR)NULL); LeaveCriticalSection(&pg_signal_crit_sec);
QueueUserAPC(pg_signal_apc, pgwin32_main_thread_handle, (ULONG_PTR) NULL);
} }
/* Signal dispatching thread */ /* Signal dispatching thread */
static DWORD WINAPI pg_signal_dispatch_thread(LPVOID param) { static DWORD WINAPI
HANDLE pipe = (HANDLE)param; pg_signal_dispatch_thread(LPVOID param)
BYTE sigNum; {
DWORD bytes; HANDLE pipe = (HANDLE) param;
BYTE sigNum;
if (!ReadFile(pipe,&sigNum,1,&bytes,NULL)) { DWORD bytes;
/* Client died before sending */
CloseHandle(pipe); if (!ReadFile(pipe, &sigNum, 1, &bytes, NULL))
return 0; {
} /* Client died before sending */
if (bytes != 1) { CloseHandle(pipe);
/* Received <bytes> bytes over signal pipe (should be 1) */ return 0;
CloseHandle(pipe); }
return 0; if (bytes != 1)
} {
WriteFile(pipe,&sigNum,1,&bytes,NULL); /* Don't care if it works or not.. */ /* Received <bytes> bytes over signal pipe (should be 1) */
FlushFileBuffers(pipe); CloseHandle(pipe);
DisconnectNamedPipe(pipe); return 0;
CloseHandle(pipe); }
WriteFile(pipe, &sigNum, 1, &bytes, NULL); /* Don't care if it works
pg_queue_signal(sigNum); * or not.. */
return 0; FlushFileBuffers(pipe);
DisconnectNamedPipe(pipe);
CloseHandle(pipe);
pg_queue_signal(sigNum);
return 0;
} }
/* Signal handling thread */ /* Signal handling thread */
static DWORD WINAPI pg_signal_thread(LPVOID param) { static DWORD WINAPI
char pipename[128]; pg_signal_thread(LPVOID param)
HANDLE pipe = INVALID_HANDLE_VALUE; {
char pipename[128];
wsprintf(pipename,"\\\\.\\pipe\\pgsignal_%i",GetCurrentProcessId()); HANDLE pipe = INVALID_HANDLE_VALUE;
for (;;) { wsprintf(pipename, "\\\\.\\pipe\\pgsignal_%i", GetCurrentProcessId());
BOOL fConnected;
HANDLE hThread; for (;;)
{
pipe = CreateNamedPipe(pipename,PIPE_ACCESS_DUPLEX, BOOL fConnected;
PIPE_TYPE_MESSAGE | PIPE_READMODE_MESSAGE | PIPE_WAIT, HANDLE hThread;
PIPE_UNLIMITED_INSTANCES,16,16,1000,NULL);
if (pipe == INVALID_HANDLE_VALUE) { pipe = CreateNamedPipe(pipename, PIPE_ACCESS_DUPLEX,
fprintf(stderr,gettext("Failed to create signal listener pipe: %i. Retrying.\n"),(int)GetLastError()); PIPE_TYPE_MESSAGE | PIPE_READMODE_MESSAGE | PIPE_WAIT,
SleepEx(500,TRUE); PIPE_UNLIMITED_INSTANCES, 16, 16, 1000, NULL);
continue; if (pipe == INVALID_HANDLE_VALUE)
} {
fprintf(stderr, gettext("Failed to create signal listener pipe: %i. Retrying.\n"), (int) GetLastError());
fConnected = ConnectNamedPipe(pipe, NULL) ? TRUE : (GetLastError() == ERROR_PIPE_CONNECTED); SleepEx(500, TRUE);
if (fConnected) { continue;
hThread = CreateThread(NULL, 0, }
(LPTHREAD_START_ROUTINE)pg_signal_dispatch_thread,
(LPVOID)pipe,0,NULL); fConnected = ConnectNamedPipe(pipe, NULL) ? TRUE : (GetLastError() == ERROR_PIPE_CONNECTED);
if (hThread == INVALID_HANDLE_VALUE) { if (fConnected)
fprintf(stderr,gettext("Failed to create signal dispatch thread: %i\n"),(int)GetLastError()); {
} hThread = CreateThread(NULL, 0,
else (LPTHREAD_START_ROUTINE) pg_signal_dispatch_thread,
CloseHandle(hThread); (LPVOID) pipe, 0, NULL);
} if (hThread == INVALID_HANDLE_VALUE)
else fprintf(stderr, gettext("Failed to create signal dispatch thread: %i\n"), (int) GetLastError());
/* Connection failed. Cleanup and try again */ else
CloseHandle(pipe); CloseHandle(hThread);
} }
return 0; else
/* Connection failed. Cleanup and try again */
CloseHandle(pipe);
}
return 0;
} }
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment