Here's the latest win32 signals code, this time in the form of a patch

against the latest shapshot. It also includes the replacement of kill()
with pqkill() and sigsetmask() with pqsigsetmask().

Passes all tests fine on my linux machine once applied. Still doesn't
link completely on Win32 - there are a few things still required. But
much closer than before.

At Bruce's request, I'm goint to write up a README file about the method
of signals delivery chosen and why the others were rejected (basically a
summary of the mailinglist discussions). I'll finish that up once/if the
patch is accepted.


Magnus Hagander

Author: bmomjian

src/backend/commands/async.c

@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/async.c,v 1.107 2004/01/07 18:56:25 neilc Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/async.c,v 1.108 2004/01/27 00:45:26 momjian Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -83,6 +83,7 @@
 #include "commands/async.h"
 #include "libpq/libpq.h"
 #include "libpq/pqformat.h"
+#include "libpq/pqsignal.h"
 #include "miscadmin.h"
 #include "storage/ipc.h"
 #include "tcop/tcopprot.h"
@@ -497,7 +498,7 @@ AtCommit_Notify(void)
 			 * for some reason.  It's OK to send the signal first, because
 			 * the other guy can't read pg_listener until we unlock it.
 			 */
-			if (kill(listenerPID, SIGUSR2) < 0)
+			if (pqkill(listenerPID, SIGUSR2) < 0)
 			{
 				/*
 				 * Get rid of pg_listener entry if it refers to a PID that

src/backend/libpq/pqsignal.c

@@ -9,7 +9,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/libpq/pqsignal.c,v 1.28 2003/11/29 19:51:49 pgsql Exp $
+ *	  $PostgreSQL: pgsql/src/backend/libpq/pqsignal.c,v 1.29 2004/01/27 00:45:26 momjian Exp $
  *
  * NOTES
  *		This shouldn't be in libpq, but the monitor and some other
@@ -38,9 +38,18 @@
  *	is to do signal-handler reinstallation, which doesn't work well
  *	at all.
  * ------------------------------------------------------------------------*/
+#ifdef WIN32
+#define WIN32_LEAN_AND_MEAN
+#define _WIN32_WINNT 0x0400
+#endif
+
 #include "postgres.h"
 
+#ifndef WIN32
 #include <signal.h>
+#else
+#include <windows.h>
+#endif
 
 #include "libpq/pqsignal.h"
 
@@ -127,6 +136,7 @@ pqinitmask(void)
 }
 
 
+#ifndef WIN32
 /*
  * Set up a signal handler
  */
@@ -149,3 +159,234 @@ pqsignal(int signo, pqsigfunc func)
 	return oact.sa_handler;
 #endif   /* !HAVE_POSIX_SIGNALS */
 }
+
+
+#else
+
+
+/* Win32 specific signals code */
+
+/* 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! */
+static CRITICAL_SECTION pg_signal_crit_sec;
+static int pg_signal_queue;
+
+#define PG_SIGNAL_COUNT 32
+static pqsigfunc pg_signal_array[PG_SIGNAL_COUNT];
+static pqsigfunc pg_signal_defaults[PG_SIGNAL_COUNT];
+static int pg_signal_mask;
+
+HANDLE pgwin32_main_thread_handle;
+
+/* Signal handling thread function */
+static DWORD WINAPI pg_signal_thread(LPVOID param);
+
+/* Initialization */
+void pgwin32_signal_initialize(void) {
+  int i;
+  HANDLE signal_thread_handle;
+  InitializeCriticalSection(&pg_signal_crit_sec);
+  
+  for (i = 0; i < PG_SIGNAL_COUNT; i++) {
+    pg_signal_array[i] = SIG_DFL;
+    pg_signal_defaults[i] = SIG_IGN;
+  }
+  pg_signal_mask = 0;
+  pg_signal_queue = 0;
+  
+  /* Get handle to main thread so we can post calls to it later */
+  if (!DuplicateHandle(GetCurrentProcess(),GetCurrentThread(),
+		       GetCurrentProcess(),&pgwin32_main_thread_handle,
+		       0,FALSE,DUPLICATE_SAME_ACCESS)) {
+    fprintf(stderr,gettext("Failed to get main thread handle!\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 
+ * Blocked signals are ignored, and will be fired at the time of
+ * the sigsetmask() call. */
+static void dispatch_queued_signals(void) {
+  int i;
+  
+  EnterCriticalSection(&pg_signal_crit_sec);
+  while (pg_signal_queue & ~pg_signal_mask) {
+    /* One or more unblocked signals queued for execution */
+    
+    int exec_mask = pg_signal_queue & ~pg_signal_mask;
+    
+    for (i = 0; i < PG_SIGNAL_COUNT; i++) {
+      if (exec_mask & sigmask(i)) {
+	/* Execute this signal */
+	pqsigfunc sig = pg_signal_array[i];
+	if (sig == SIG_DFL)
+	  sig = pg_signal_defaults[i];
+	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);
+}
+
+/* signal masking. Only called on main thread, no sync required */
+int pqsigsetmask(int mask) {
+  int prevmask;
+  prevmask = pg_signal_mask;
+  pg_signal_mask = mask;
+  
+  /* 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 */
+pqsigfunc pqsignal(int signum, pqsigfunc handler) {
+  pqsigfunc prevfunc;
+  if (signum >= PG_SIGNAL_COUNT || signum < 0)
+    return SIG_ERR;
+  prevfunc = pg_signal_array[signum];
+  pg_signal_array[signum] = handler;
+  return prevfunc;
+}
+
+/* signal sending */
+int pqkill(int pid, int sig) {
+  char pipename[128];
+  BYTE sigData = sig;
+  BYTE sigRet = 0;
+  DWORD bytes;
+  
+  if (sig >= PG_SIGNAL_COUNT || sig <= 0) {
+    errno = EINVAL;
+    return -1;
+  }
+  if (pid <= 0) {
+    /* No support for process groups */
+    errno = EINVAL;
+    return -1;
+  }
+  wsprintf(pipename,"\\\\.\\pipe\\pgsignal_%i",pid);
+  if (!CallNamedPipe(pipename,&sigData,1,&sigRet,1,&bytes,1000)) {
+    if (GetLastError() == ERROR_FILE_NOT_FOUND)
+      errno = ESRCH;
+    else if (GetLastError() == ERROR_ACCESS_DENIED)
+      errno = EPERM;
+    else
+      errno = EINVAL;
+    return -1;
+  }
+  if (bytes != 1 || sigRet != sig) {
+    errno = ESRCH;
+    return -1;
+  }
+  
+  return 0;
+}
+
+/* APC callback scheduled on main thread when signals are fired */
+static void CALLBACK pg_signal_apc(ULONG_PTR param) {
+  dispatch_queued_signals();
+}
+
+/*
+ * All functions below execute on the signal handler thread
+ * and must be synchronized as such!
+ * NOTE! The only global variable that can be used is
+ * pg_signal_queue!
+ */
+
+
+static void pg_queue_signal(int signum) {
+  if (signum >= PG_SIGNAL_COUNT || signum < 0)
+    return;
+  
+  EnterCriticalSection(&pg_signal_crit_sec);
+  pg_signal_queue |= sigmask(signum);
+  LeaveCriticalSection(&pg_signal_crit_sec);
+  
+  QueueUserAPC(pg_signal_apc,pgwin32_main_thread_handle,(ULONG_PTR)NULL);
+}
+
+/* Signal dispatching thread */
+static DWORD WINAPI pg_signal_dispatch_thread(LPVOID param) {
+  HANDLE pipe = (HANDLE)param;
+  BYTE sigNum;
+  DWORD bytes;
+  
+  if (!ReadFile(pipe,&sigNum,1,&bytes,NULL)) {
+    /* Client died before sending */
+    CloseHandle(pipe);
+    return 0;
+  }
+  if (bytes != 1) {
+    /* Received <bytes> bytes over signal pipe (should be 1) */
+    CloseHandle(pipe);
+    return 0;
+  }
+  WriteFile(pipe,&sigNum,1,&bytes,NULL); /* Don't care if it works or not.. */
+  FlushFileBuffers(pipe);
+  DisconnectNamedPipe(pipe);
+  CloseHandle(pipe);
+  
+  pg_queue_signal(sigNum);
+  return 0;
+}
+
+/* Signal handling thread */
+static DWORD WINAPI pg_signal_thread(LPVOID param) {
+  char pipename[128];
+  HANDLE pipe = INVALID_HANDLE_VALUE;
+  
+  wsprintf(pipename,"\\\\.\\pipe\\pgsignal_%i",GetCurrentProcessId());
+  
+  for (;;) {
+    BOOL fConnected;
+    HANDLE hThread;
+    
+    pipe = CreateNamedPipe(pipename,PIPE_ACCESS_DUPLEX,
+			   PIPE_TYPE_MESSAGE | PIPE_READMODE_MESSAGE | PIPE_WAIT,
+			   PIPE_UNLIMITED_INSTANCES,16,16,1000,NULL);
+    if (pipe == INVALID_HANDLE_VALUE) {
+      fprintf(stderr,gettext("Failed to create signal listener pipe: %i. Retrying.\n"),(int)GetLastError());
+      SleepEx(500,TRUE);
+      continue;
+    }
+    
+    fConnected = ConnectNamedPipe(pipe, NULL) ? TRUE : (GetLastError() == ERROR_PIPE_CONNECTED);
+    if (fConnected) {
+      hThread = CreateThread(NULL, 0, 
+			     (LPTHREAD_START_ROUTINE)pg_signal_dispatch_thread,
+			     (LPVOID)pipe,0,NULL);
+      if (hThread == INVALID_HANDLE_VALUE) {
+	fprintf(stderr,gettext("Failed to create signal dispatch thread: %i\n"),(int)GetLastError());
+      }
+      else 
+	CloseHandle(hThread);
+    }
+    else
+      /* Connection failed. Cleanup and try again */
+      CloseHandle(pipe);
+  }
+  return 0;
+}
+
+
+#endif

src/backend/main/main.c

@@ -13,7 +13,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/main/main.c,v 1.71 2004/01/11 03:49:31 momjian Exp $
+ *	  $PostgreSQL: pgsql/src/backend/main/main.c,v 1.72 2004/01/27 00:45:26 momjian Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -40,6 +40,9 @@
 #include "utils/help_config.h"
 #include "utils/ps_status.h"
 #include "pgstat.h"
+#ifdef WIN32
+#include "libpq/pqsignal.h"
+#endif
 
 
 
@@ -97,6 +100,8 @@ main(int argc, char *argv[])
 					argv[0], err);
 			exit(1);
 		}
+
+		pgwin32_signal_initialize();
 	}
 #endif
 

src/backend/port/sysv_sema.c

@@ -8,7 +8,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/port/sysv_sema.c,v 1.12 2003/12/01 22:15:37 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/port/sysv_sema.c,v 1.13 2004/01/27 00:45:26 momjian Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -31,6 +31,7 @@
 #include "miscadmin.h"
 #include "storage/ipc.h"
 #include "storage/pg_sema.h"
+#include "libpq/pqsignal.h"
 
 
 #ifndef HAVE_UNION_SEMUN
@@ -232,7 +233,7 @@ IpcSemaphoreCreate(int numSems)
 			continue;			/* oops, GETPID failed */
 		if (creatorPID != getpid())
 		{
-			if (kill(creatorPID, 0) == 0 ||
+			if (pqkill(creatorPID, 0) == 0 ||
 				errno != ESRCH)
 				continue;		/* sema belongs to a live process */
 		}

src/backend/port/sysv_shmem.c

@@ -10,7 +10,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/port/sysv_shmem.c,v 1.28 2004/01/07 18:56:27 neilc Exp $
+ *	  $PostgreSQL: pgsql/src/backend/port/sysv_shmem.c,v 1.29 2004/01/27 00:45:26 momjian Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -33,6 +33,7 @@
 #include "miscadmin.h"
 #include "storage/ipc.h"
 #include "storage/pg_shmem.h"
+#include "libpq/pqsignal.h"
 
 
 typedef key_t IpcMemoryKey;		/* shared memory key passed to shmget(2) */
@@ -284,7 +285,7 @@ PGSharedMemoryCreate(uint32 size, bool makePrivate, int port)
 		hdr = (PGShmemHeader *) memAddress;
 		if (hdr->creatorPID != getpid())
 		{
-			if (kill(hdr->creatorPID, 0) == 0 || errno != ESRCH)
+			if (pqkill(hdr->creatorPID, 0) == 0 || errno != ESRCH)
 			{
 				shmdt(memAddress);
 				continue;		/* segment belongs to a live process */