Fix handle

Author: ildus

jsonbc.c

@@ -157,18 +157,23 @@ jsonbc_shmem_startup_hook(void)
 
 		for (i = 0; i < jsonbc_nworkers; i++)
 		{
+			size_t queue_size = jsonbc_get_queue_size();
+
 			jsonbc_shm_worker	*wd = shm_toc_allocate(toc, sizeof(jsonbc_shm_worker));
 
 			/* each worker will have two mq, for input and output */
-			wd->mqin = shm_mq_create(shm_toc_allocate(toc, jsonbc_get_queue_size()),
-							   jsonbc_get_queue_size());
-			wd->mqout = shm_mq_create(shm_toc_allocate(toc, jsonbc_get_queue_size()),
-							   jsonbc_get_queue_size());
+			wd->mqin = shm_mq_create(shm_toc_allocate(toc, queue_size), queue_size);
+			wd->mqout = shm_mq_create(shm_toc_allocate(toc, queue_size), queue_size);
 
 			/* init worker context */
 			pg_atomic_init_flag(&wd->busy);
 			wd->proc = NULL;
 
+			shm_mq_clean_receiver(wd->mqin);
+			shm_mq_clean_receiver(wd->mqout);
+			shm_mq_clean_sender(wd->mqin);
+			shm_mq_clean_sender(wd->mqout);
+
 			shm_toc_insert(toc, i + 1, wd);
 			shm_toc_insert(toc, mqkey++, wd->mqin);
 			shm_toc_insert(toc, mqkey++, wd->mqout);

jsonbc_worker.c

@@ -338,6 +338,7 @@ void
 worker_main(Datum arg)
 {
 	MemoryContext	worker_context;
+	shm_mq_handle  *mqh = NULL;
 
 	/* Establish signal handlers before unblocking signals */
 	pqsignal(SIGTERM, handle_sigterm);
@@ -365,10 +366,11 @@ worker_main(Datum arg)
 		Size	nbytes;
 		void   *data;
 
-		shm_mq_handle  *mqh;
 		shm_mq_result	resmq;
 
-		mqh = shm_mq_attach(worker_state->mqin, NULL, NULL);
+		if (!mqh)
+			mqh = shm_mq_attach(worker_state->mqin, NULL, NULL);
+
 		resmq = shm_mq_receive(mqh, &nbytes, &data, true);
 
 		if (resmq == SHM_MQ_SUCCESS)
@@ -409,7 +411,10 @@ worker_main(Datum arg)
 				elog(NOTICE, "jsonbc: backend detached early");
 
 			shm_mq_detach(mqh);
-			pfree((void *) iov.data);
+			MemoryContextReset(worker_context);
+
+			/* mark we need new handle */
+			mqh = NULL;
 		}
 
 		if (shutdown_requested)