Use semaphores in backends, fix mq bug

Author: ildus

jsonbc.c

@@ -150,6 +150,7 @@ jsonbc_shmem_startup_hook(void)
 		toc = shm_toc_create(JSONBC_SHM_MQ_MAGIC, workers_data, size);
 		hdr = shm_toc_allocate(toc, sizeof(jsonbc_shm_hdr));
 		hdr->workers_ready = 0;
+		sem_init(&hdr->workers_sem, 1, jsonbc_nworkers);
 		shm_toc_insert(toc, 0, hdr);
 		mqkey = jsonbc_nworkers + 1;
 
@@ -322,6 +323,7 @@ jsonbc_communicate(shm_mq_iovec *iov, int iov_len,
 	bool				callback_succeded = false;
 	shm_mq_result		resmq;
 	shm_mq_handle	   *mqh;
+	jsonbc_shm_hdr	   *hdr;
 
 	char			   *res;
 	Size				reslen;
@@ -330,7 +332,9 @@ jsonbc_communicate(shm_mq_iovec *iov, int iov_len,
 		/* TODO: maybe add support of multiple databases for dictionaries */
 		elog(ERROR, "jsonbc workers are not available");
 
-again:
+	hdr = shm_toc_lookup(toc, 0, false);
+	sem_wait(&hdr->workers_sem);
+
 	for (i = 0; i < jsonbc_nworkers; i++)
 	{
 		jsonbc_shm_worker *wd = shm_toc_lookup(toc, i + 1, false);
@@ -339,6 +343,8 @@ jsonbc_communicate(shm_mq_iovec *iov, int iov_len,
 
 		/* send data */
 		shm_mq_set_sender(wd->mqin, MyProc);
+		shm_mq_set_receiver(wd->mqout, MyProc);
+
 		mqh = shm_mq_attach(wd->mqin, NULL, NULL);
 		resmq = shm_mq_sendv(mqh, iov, iov_len, false);
 		if (resmq != SHM_MQ_SUCCESS)
@@ -348,7 +354,6 @@ jsonbc_communicate(shm_mq_iovec *iov, int iov_len,
 		/* get data */
 		if (!detached)
 		{
-			shm_mq_set_receiver(wd->mqout, MyProc);
 			mqh = shm_mq_attach(wd->mqout, NULL, NULL);
 			resmq = shm_mq_receive(mqh, &reslen, (void **) &res, false);
 			if (resmq != SHM_MQ_SUCCESS)
@@ -361,6 +366,7 @@ jsonbc_communicate(shm_mq_iovec *iov, int iov_len,
 		}
 
 		/* clean self as receiver and unlock mq */
+		shm_mq_clean_sender(wd->mqin);
 		shm_mq_clean_receiver(wd->mqout);
 		pg_atomic_clear_flag(&wd->busy);
 
@@ -371,14 +377,10 @@ jsonbc_communicate(shm_mq_iovec *iov, int iov_len,
 			elog(ERROR, "jsonbc: communication error");
 
 		/* we're done here */
-		return;
+		break;
 	}
 
-	CHECK_FOR_INTERRUPTS();
-	pg_usleep(100);
-
-	/* TODO: add attempts count check */
-	goto again;
+	sem_post(&hdr->workers_sem);
 }
 
 /* Get key IDs using workers */

jsonbc.h

@@ -2,6 +2,7 @@
 #define JSONBC_H
 
 #include <postgres.h>
+#include <semaphore.h>
 
 #include "port/atomics.h"
 #include "storage/proc.h"
@@ -17,7 +18,8 @@ typedef enum {
 
 typedef struct jsonbc_shm_hdr
 {
-	volatile int workers_ready;
+	sem_t			workers_sem;
+	volatile int	workers_ready;
 } jsonbc_shm_hdr;
 
 typedef struct jsonbc_shm_worker

jsonbc_worker.c

@@ -250,6 +250,8 @@ jsonbc_get_key_ids_slow(Oid cmoptoid, char *buf, uint32 *idsbuf, int nkeys)
 		if (SPI_exec(sql, 0) != SPI_OK_SELECT)
 			elog(ERROR, "SPI_exec failed");
 
+		pfree(sql);
+
 		if (SPI_processed == 0)
 		{
 			char *sql2 = psprintf("with t as (select (coalesce(max(id), 0) + 1) new_id from "
@@ -259,6 +261,8 @@ jsonbc_get_key_ids_slow(Oid cmoptoid, char *buf, uint32 *idsbuf, int nkeys)
 
 			if (SPI_exec(sql2, 0) != SPI_OK_INSERT_RETURNING)
 				elog(ERROR, "SPI_exec failed");
+
+			pfree(sql2);
 		}
 
 		datum = SPI_getbinval(SPI_tuptable->vals[0],
@@ -275,7 +279,6 @@ jsonbc_get_key_ids_slow(Oid cmoptoid, char *buf, uint32 *idsbuf, int nkeys)
 			buf++;
 
 		buf++;
-		pfree(sql);
 	}
 	SPI_finish();
 	relation_close(rel, ShareLock);
@@ -382,7 +385,7 @@ worker_main(Datum arg)
 			JsonbcCommand	cmd;
 			Oid				cmoptoid;
 			shm_mq_iovec	iov;
-			char			*ptr = data;
+			char		   *ptr = data;
 			int				nkeys = *((int *) ptr);
 
 			ptr += sizeof(int);
@@ -409,7 +412,6 @@ worker_main(Datum arg)
 			}
 
 			shm_mq_detach(mqh);
-			shm_mq_clean_sender(worker_state->mqin);
 
 			mqh = shm_mq_attach(worker_state->mqout, NULL, NULL);
 			resmq = shm_mq_sendv(mqh, &iov, 1, false);