fixup! Fix compilation

Author: ildus

jsonbd.c

@@ -60,7 +60,7 @@ static uint32 decode_varbyte(unsigned char *ptr);
 static size_t
 jsonbd_get_queue_size(void)
 {
-	return (Size) (jsonbd_queue_size * 1024);
+	return (Size) (shm_mq_minimum_size + jsonbd_queue_size * 1024);
 }
 
 static size_t
@@ -74,7 +74,9 @@ jsonbd_shmem_size(void)
 	shm_toc_initialize_estimator(&e);
 
 	shm_toc_estimate_chunk(&e, sizeof(jsonbd_shm_hdr));
-	shm_toc_estimate_chunk(&e, PGSemaphoreShmemSize(1));
+
+	/* two queues for launcher */
+	shm_toc_estimate_chunk(&e, shm_mq_minimum_size * 2);
 
 	for (i = 0; i < MAX_JSONBD_WORKERS; i++)
 	{
@@ -83,8 +85,8 @@ jsonbd_shmem_size(void)
 		shm_toc_estimate_chunk(&e, jsonbd_get_queue_size());
 	}
 
-	/* 3 keys each worker + 3 for header (header itself, launcher and its two queues) */
-	shm_toc_estimate_keys(&e, MAX_JSONBD_WORKERS * 3 + 4);
+	/* 3 keys each worker + 3 for header (header itself and two queues) */
+	shm_toc_estimate_keys(&e, MAX_JSONBD_WORKERS * 3 + 3);
 	size = shm_toc_estimate(&e);
 	return size;
 }
@@ -95,14 +97,13 @@ jsonbd_shmem_size(void)
  * About keys in toc:
  *	0 - for header
  *	1..MAX_JSONBD_WORKERS - workers
- *	MAX_JSONBD_WORKERS + 1 - launcher
- *	MAX_JSONBD_WORKERS + 2 .. - queues
+ *	MAX_JSONBD_WORKERS + 1 .. - queues
  */
 static void
 jsonbd_init_worker(shm_toc *toc, jsonbd_shm_worker *wd, int worker_num,
 		size_t queue_size)
 {
-	static int mqkey = MAX_JSONBD_WORKERS + 2;
+	static int mqkey = MAX_JSONBD_WORKERS + 1;
 
 	/* each worker will have two mq, for input and output */
 	wd->mqin = shm_mq_create(shm_toc_allocate(toc, queue_size), queue_size);
@@ -146,15 +147,11 @@ jsonbd_shmem_startup_hook(void)
 
 		toc = shm_toc_create(JSONBD_SHM_MQ_MAGIC, workers_data, size);
 
-		/* initialize header */
+		/* Initialize header */
 		hdr = shm_toc_allocate(toc, sizeof(jsonbd_shm_hdr));
 		hdr->workers_ready = 0;
-		hdr->launcher_sem = PGSemaphoreCreate();
-		jsonbd_init_worker(toc, &hdr->launcher, MAX_JSONBD_WORKERS + 1, sizeof(Oid));
-
-		for (i = 0; i < MAX_DATABASES; i++)
-			sem_init(&hdr->workers_sem[i], 1, jsonbd_nworkers);
-
+		sem_init(&hdr->launcher_sem, 1, 1);
+		jsonbd_init_worker(toc, &hdr->launcher, 0, shm_mq_minimum_size);
 		shm_toc_insert(toc, 0, hdr);
 
 		for (i = 0; i < MAX_JSONBD_WORKERS; i++)
@@ -285,6 +282,7 @@ jsonbd_communicate(shm_mq_iovec *iov, int iov_len,
 	int					i,
 						j;
 	bool				detached = false;
+	bool				launch_failed = false;
 	bool				callback_succeded = false;
 	shm_mq_result		resmq;
 	shm_mq_handle	   *mqh;
@@ -321,8 +319,8 @@ jsonbd_communicate(shm_mq_iovec *iov, int iov_len,
 				wd = shm_toc_lookup(toc, j + 1, false);
 
 				if (wd->dboid != MyDatabaseId)
-					/* somehow not all workers started for this database */
-					break;
+					/* somehow not all workers started for this database, try next */
+					continue;
 
 				if (pg_atomic_test_set_flag(&wd->busy))
 					goto comm;
@@ -337,7 +335,7 @@ jsonbd_communicate(shm_mq_iovec *iov, int iov_len,
 		 * there are no workers for our database,
 		 * so we should launch them using our jsonbd workers launcher
 		 */
-		PGSemaphoreLock(hdr->launcher_sem);
+		sem_wait(&hdr->launcher_sem);
 		shm_mq_set_sender(hdr->launcher.mqin, MyProc);
 		shm_mq_set_receiver(hdr->launcher.mqout, MyProc);
 
@@ -355,15 +353,21 @@ jsonbd_communicate(shm_mq_iovec *iov, int iov_len,
 			if (resmq != SHM_MQ_SUCCESS)
 				detached = true;
 
+			if (reslen != 2 || res[0] == 'n')
+				launch_failed = true;
+
 			shm_mq_detach(mqh);
 		}
 
 		shm_mq_clean_sender(hdr->launcher.mqin);
 		shm_mq_clean_receiver(hdr->launcher.mqout);
-		PGSemaphoreUnlock(hdr->launcher_sem);
+		sem_post(&hdr->launcher_sem);
 
 		if (detached)
 			elog(ERROR, "jsonbd: workers launcher was detached");
+
+		if (launch_failed)
+			elog(ERROR, "jsonbd: could not launch dictionary worker, see logs");
 	}
 
 comm:

jsonbd.h

@@ -8,7 +8,6 @@
 #include "port/atomics.h"
 #include "storage/proc.h"
 #include "storage/shm_mq.h"
-#include "storage/pg_sema.h"
 
 #define JSONBD_SHM_MQ_MAGIC		0xAAAA
 
@@ -34,7 +33,7 @@ typedef struct jsonbd_shm_worker
 /* Shared memory structures */
 typedef struct jsonbd_shm_hdr
 {
-	PGSemaphore			launcher_sem;
+	sem_t				launcher_sem;
 	sem_t				workers_sem[MAX_DATABASES];
 	volatile int		workers_ready;
 	jsonbd_shm_worker	launcher;

jsonbd_worker.c

@@ -16,6 +16,7 @@
 #include "catalog/pg_compression_opt.h"
 #include "catalog/pg_extension.h"
 #include "commands/extension.h"
+#include "commands/dbcommands.h"
 #include "executor/spi.h"
 #include "port/atomics.h"
 #include "storage/ipc.h"
@@ -525,6 +526,7 @@ jsonbd_worker_launcher(void)
 
 	shm_mq_handle  *mqh = NULL;
 	int				worker_num	= 1;
+	int				database_num = 0;
 
 	/* Establish signal handlers before unblocking signals */
 	pqsignal(SIGTERM, handle_sigterm);
@@ -544,6 +546,8 @@ jsonbd_worker_launcher(void)
 	shm_mq_set_receiver(worker_state->mqin, MyProc);
 	shm_mq_set_sender(worker_state->mqout, MyProc);
 
+	elog(LOG, "jsonbd launcher started with pid: %d", MyProcPid);
+
 	while (true)
 	{
 		int		rc;
@@ -559,28 +563,42 @@ jsonbd_worker_launcher(void)
 
 		if (resmq == SHM_MQ_SUCCESS)
 		{
-			bool	success = true;
+			int		started = 0;
 			int		i;
 			Oid		dboid;
 
-			Assert(nbytes == sizeof(Oid));
-			dboid = *((Oid *) data);
-
-			/* start workers for specified database */
-			for (i=0; i < jsonbd_nworkers; i++)
+			if (database_num >= MAX_DATABASES)
+				elog(NOTICE, "jsonbd: reached maximum count of supported databases");
+			else
 			{
-				bool res;
-				res = jsonbd_register_worker(worker_num++, dboid);
-				if (!res)
-					success = false;
+				Assert(nbytes == sizeof(Oid));
+				dboid = *((Oid *) data);
+
+				/* start workers for specified database */
+				for (i=0; i < jsonbd_nworkers; i++)
+				{
+					bool res;
+					res = jsonbd_register_worker(worker_num++, dboid);
+					if (res)
+						started++;
+				}
 			}
 
 			shm_mq_detach(mqh);
 			mqh = shm_mq_attach(worker_state->mqout, NULL, NULL);
-			if (success)
-				resmq = shm_mq_sendv(mqh, &((shm_mq_iovec) {"ok", 3}), 1, false);
+			if (started)
+			{
+				if (started != jsonbd_nworkers)
+					elog(NOTICE, "jsonbd: not all workers for %d has started", dboid);
+
+				/* semaphore value should be equal to workers count for database */
+				sem_init(&hdr->workers_sem[database_num], 1, started);
+
+				/* we report ok if only one worker has started */
+				resmq = shm_mq_sendv(mqh, &((shm_mq_iovec) {"y", 2}), 1, false);
+			}
 			else
-				resmq = shm_mq_sendv(mqh, &((shm_mq_iovec) {"fail", 5}), 1, false);
+				resmq = shm_mq_sendv(mqh, &((shm_mq_iovec) {"n", 2}), 1, false);
 
 			if (resmq != SHM_MQ_SUCCESS)
 				elog(NOTICE, "jsonbd: backend detached early");
@@ -754,6 +772,12 @@ jsonbd_register_worker(int worker_num, Oid dboid)
 	BackgroundWorkerHandle *bgw_handle;
 	jsonbd_worker_args *worker_args;
 
+	if (worker_num > MAX_JSONBD_WORKERS)
+	{
+		elog(LOG, "Reached maximum count of jsonbd dictionary workers");
+		return false;
+	}
+
 	/* Initialize DSM segment */
 	seg = dsm_create(sizeof(jsonbd_worker_args), 0);
 	worker_args = (jsonbd_worker_args *) dsm_segment_address(seg);
@@ -767,7 +791,13 @@ jsonbd_register_worker(int worker_num, Oid dboid)
 	worker.bgw_notify_pid = MyProcPid;
 	memcpy(worker.bgw_library_name, "jsonbd", BGW_MAXLEN);
 	memcpy(worker.bgw_function_name, CppAsString(jsonbd_worker_main), BGW_MAXLEN);
-	snprintf(worker.bgw_name, BGW_MAXLEN, "jsonbd dictionary worker %d", worker_num);
+
+	/* we need transaction to access syscache */
+	start_xact_command();
+	snprintf(worker.bgw_name, BGW_MAXLEN, "jsonbd, worker %d, db: \"%s\"",
+			worker_num, get_database_name(dboid));
+	finish_xact_command();
+
 	worker.bgw_main_arg = UInt32GetDatum(dsm_segment_handle(seg));
 
 	/* Start dynamic worker */