gpcontrib/orafce/pipe.c

#include "postgres.h"
#include "funcapi.h"
#include "fmgr.h"
#include "access/htup_details.h"
#include "storage/shmem.h"
#include "utils/memutils.h"
#include "utils/timestamp.h"
#include "storage/lwlock.h"
#include "miscadmin.h"
#include "lib/stringinfo.h"
#include "catalog/pg_type.h"
#include "utils/builtins.h"
#include "utils/date.h"
#include "utils/numeric.h"

#if PG_VERSION_NUM >= 140000

#include "utils/wait_event.h"

#elif PG_VERSION_NUM >= 130000

#include "pgstat.h"

#endif

#include "shmmc.h"
#include "pipe.h"
#include "orafce.h"
#include "builtins.h"

#include <string.h>

/*
 * @ Pavel Stehule 2006-2023
 */

#ifndef _GetCurrentTimestamp
#define _GetCurrentTimestamp()	GetCurrentTimestamp()
#endif

#ifndef GetNowFloat
#ifdef HAVE_INT64_TIMESTAMP
#define GetNowFloat()   ((float8) _GetCurrentTimestamp() / 1000000.0)
#else
#define GetNowFloat()   _GetCurrentTimestamp()
#endif
#endif

#define RESULT_DATA			0
#define RESULT_TIMEOUT		1

/* in sec 1000 days */
#define MAXWAIT		86400000


PG_FUNCTION_INFO_V1(dbms_pipe_pack_message_text);
PG_FUNCTION_INFO_V1(dbms_pipe_unpack_message_text);
PG_FUNCTION_INFO_V1(dbms_pipe_send_message);
PG_FUNCTION_INFO_V1(dbms_pipe_receive_message);
PG_FUNCTION_INFO_V1(dbms_pipe_unique_session_name);
PG_FUNCTION_INFO_V1(dbms_pipe_list_pipes);
PG_FUNCTION_INFO_V1(dbms_pipe_next_item_type);
PG_FUNCTION_INFO_V1(dbms_pipe_create_pipe);
PG_FUNCTION_INFO_V1(dbms_pipe_create_pipe_2);
PG_FUNCTION_INFO_V1(dbms_pipe_create_pipe_1);
PG_FUNCTION_INFO_V1(dbms_pipe_reset_buffer);
PG_FUNCTION_INFO_V1(dbms_pipe_purge);
PG_FUNCTION_INFO_V1(dbms_pipe_remove_pipe);
PG_FUNCTION_INFO_V1(dbms_pipe_pack_message_date);
PG_FUNCTION_INFO_V1(dbms_pipe_unpack_message_date);
PG_FUNCTION_INFO_V1(dbms_pipe_pack_message_timestamp);
PG_FUNCTION_INFO_V1(dbms_pipe_unpack_message_timestamp);
PG_FUNCTION_INFO_V1(dbms_pipe_pack_message_number);
PG_FUNCTION_INFO_V1(dbms_pipe_unpack_message_number);
PG_FUNCTION_INFO_V1(dbms_pipe_pack_message_bytea);
PG_FUNCTION_INFO_V1(dbms_pipe_unpack_message_bytea);
PG_FUNCTION_INFO_V1(dbms_pipe_pack_message_record);
PG_FUNCTION_INFO_V1(dbms_pipe_unpack_message_record);
PG_FUNCTION_INFO_V1(dbms_pipe_pack_message_integer);
PG_FUNCTION_INFO_V1(dbms_pipe_pack_message_bigint);

typedef enum {
	IT_NO_MORE_ITEMS = 0,
	IT_NUMBER = 9,
	IT_VARCHAR = 11,
	IT_DATE = 12,
	IT_TIMESTAMPTZ = 13,
	IT_BYTEA = 23,
	IT_RECORD = 24
} message_data_type;

typedef struct _queue_item {
	void *ptr;
	struct _queue_item *next_item;
} queue_item;

typedef struct {
	long		identity;
	bool		is_valid;
	bool		registered;
	char	   *pipe_name;
	char	   *creator;
	Oid			uid;
	struct _queue_item *items;
	struct _queue_item *last_item;
	int16		count;
	int16		limit;
	int			size;
} orafce_pipe;

typedef struct {
	int32 size;
	message_data_type type;
	Oid tupType;
} message_data_item;

typedef struct {
	int32 size;
	int32 items_count;
	message_data_item *next;
} message_buffer;

#define message_buffer_size		(MAXALIGN(sizeof(message_buffer)))
#define message_buffer_get_content(buf)	((message_data_item *) (((char*)buf)+message_buffer_size))


#define message_data_item_size	(MAXALIGN(sizeof(message_data_item)))
#define message_data_get_content(msg) (((char *)msg) + message_data_item_size)
#define message_data_item_next(msg) \
	((message_data_item *) (message_data_get_content(msg) + MAXALIGN(msg->size)))

typedef struct PipesFctx {
	int pipe_nth;
} PipesFctx;

typedef struct
{
	int			tranche_id;
	LWLock		shmem_lock;

	orafce_pipe *pipes;
	alert_event *events;
	alert_lock *locks;

#if PG_VERSION_NUM >= 130000

	ConditionVariable pipe_cv;
	ConditionVariable alert_cv;

#endif

	size_t		size;
	int			sid;
	long		identity_seq;
	vardata		data[1]; /* flexible array member */
} sh_memory;

#define sh_memory_size			(offsetof(sh_memory, data))

message_buffer *output_buffer = NULL;
message_buffer *input_buffer = NULL;

orafce_pipe* pipes = NULL;

long	   *identity_seq = NULL;

#define NOT_INITIALIZED		NULL

LWLockId shmem_lockid = NOT_INITIALIZED;

int sid;                                 /* session id */

extern alert_event *events;
extern alert_lock  *locks;

#if PG_VERSION_NUM >= 130000

ConditionVariable *pipe_cv = NULL;
ConditionVariable *alert_cv = NULL;

#endif

/*
 * write on writer size bytes from ptr
 */
static void
pack_field(message_buffer *buffer, message_data_type type,
			int32 size, void *ptr, Oid tupType)
{
	int			len;
	message_data_item *message;

	len = MAXALIGN(size) + message_data_item_size;
	if (MAXALIGN(buffer->size) + len > LOCALMSGSZ - message_buffer_size)
		ereport(ERROR,
				(errcode(ERRCODE_OUT_OF_MEMORY),
				 errmsg("out of memory"),
				 errdetail("Packed message is bigger than local buffer."),
				 errhint("Increase LOCALMSGSZ in 'pipe.h' and recompile library.")));

	if (buffer->next == NULL)
		buffer->next =  message_buffer_get_content(buffer);

	message = buffer->next;

	message->size = size;
	message->type = type;
	message->tupType = tupType;

	/* padding bytes have to be zeroed - buffer creator is responsible to clear memory */

	memcpy(message_data_get_content(message), ptr, size);

	buffer->size += len;
	buffer->items_count++;
	buffer->next = message_data_item_next(message);
}

static void*
unpack_field(message_buffer *buffer, message_data_type *type,
				int32 *size, Oid *tupType)
{
	void	   *ptr;
	message_data_item *message;

	Assert(buffer);
	Assert(buffer->items_count > 0);
	Assert(buffer->next);

	message = buffer->next;

	Assert(message);

	*size = message->size;
	*type = message->type;
	*tupType = message->tupType;
	ptr = message_data_get_content(message);

	buffer->next = --buffer->items_count > 0 ? message_data_item_next(message) : NULL;

	return ptr;
}

/*
 * Add ptr to queue. If pipe doesn't exist, register new pipe
 */
bool
ora_lock_shmem(size_t size, int max_pipes, int max_events, int max_locks, bool reset)
{
	bool		found;

	/* reset is always false, really */
	Assert(!reset);

	if (pipes == NULL)
	{
		sh_memory  *sh_mem;

		LWLockAcquire(AddinShmemInitLock, LW_EXCLUSIVE);
		sh_mem = ShmemInitStruct("dbms_pipe", size, &found);
		if (!found)
		{
			int			i;

			sh_mem->tranche_id = LWLockNewTrancheId();
			LWLockInitialize(&sh_mem->shmem_lock, sh_mem->tranche_id);

			LWLockRegisterTranche(sh_mem->tranche_id, "orafce");
			shmem_lockid = &sh_mem->shmem_lock;

			sh_mem->identity_seq = 0;

			sh_mem->size = size - sh_memory_size;
			ora_sinit(sh_mem->data, size, true);
			pipes = sh_mem->pipes = ora_salloc(max_pipes*sizeof(orafce_pipe));
			sid = sh_mem->sid = 1;

			for (i = 0; i < max_pipes; i++)
				pipes[i].is_valid = false;

			events = sh_mem->events = ora_salloc(max_events*sizeof(alert_event));
			locks = sh_mem->locks = ora_salloc(max_locks*sizeof(alert_lock));

			for (i = 0; i < max_events; i++)
			{
				events[i].event_name = NULL;
				events[i].max_receivers = 0;
				events[i].receivers = NULL;
				events[i].messages = NULL;
			}
			for (i = 0; i < max_locks; i++)
			{
				locks[i].sid = -1;
				locks[i].echo = NULL;
			}

#if PG_VERSION_NUM >= 130000

			ConditionVariableInit(&sh_mem->pipe_cv);
			ConditionVariableInit(&sh_mem->alert_cv);

			pipe_cv = &sh_mem->pipe_cv;
			alert_cv = &sh_mem->alert_cv;

#endif

			identity_seq = &sh_mem->identity_seq;
		}
		else
		{
			LWLockRegisterTranche(sh_mem->tranche_id, "orafce");
			shmem_lockid = &sh_mem->shmem_lock;

#if PG_VERSION_NUM >= 130000

			pipe_cv = &sh_mem->pipe_cv;
			alert_cv = &sh_mem->alert_cv;

#endif

			pipes = sh_mem->pipes;
			ora_sinit(sh_mem->data, sh_mem->size, false);
			sid = ++(sh_mem->sid);
			events = sh_mem->events;
			locks = sh_mem->locks;

			identity_seq = &sh_mem->identity_seq;
		}

		LWLockRelease(AddinShmemInitLock);
	}

	Assert(pipes != NULL);
	LWLockAcquire(shmem_lockid, LW_EXCLUSIVE);
	return true;
}

#define		NOT_ASSIGNED_IDENTITY			-1

/*
 * can be enhanced access/hash.h
 */
static orafce_pipe*
find_pipe(text* pipe_name,
		  bool* created,
		  bool only_check,
		  long *expected_identity,
		  bool *identity_alarm)
{
	int			i;
	orafce_pipe *result = NULL;

	*created = false;

	Assert(!expected_identity || identity_alarm);

	if (identity_alarm)
		*identity_alarm = false;

	for (i = 0; i < MAX_PIPES; i++)
	{
		if (pipes[i].is_valid &&
			strncmp((char*)VARDATA(pipe_name), pipes[i].pipe_name, VARSIZE(pipe_name) - VARHDRSZ) == 0
			&& (strlen(pipes[i].pipe_name) == (VARSIZE(pipe_name) - VARHDRSZ)))
		{
			if (expected_identity && *expected_identity >= 0
				&& pipes[i].identity != *expected_identity)
			{
				*identity_alarm = true;
				return result;
			}

			/* check owner if non public pipe */
			if (pipes[i].creator != NULL && pipes[i].uid != GetUserId())
			{
				ereport(ERROR,
						(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
						 errmsg("insufficient privilege"),
						 errdetail("Insufficient privilege to access pipe")));
			}

			if (expected_identity)
				*expected_identity = pipes[i].identity;

			return &pipes[i];
		}
	}

	if (only_check)
		return result;

	if (expected_identity && *expected_identity >= 0)
	{
		*identity_alarm = true;
		return result;
	}

	for (i = 0; i < MAX_PIPES; i++)
		if (!pipes[i].is_valid)
		{
			if (NULL != (pipes[i].pipe_name = ora_scstring(pipe_name)))
			{
				pipes[i].is_valid = true;
				pipes[i].registered = false;
				pipes[i].creator = NULL;
				pipes[i].uid = -1;
				pipes[i].count = 0;
				pipes[i].limit = -1;

				*created = true;
				result = &pipes[i];

				if (expected_identity)
					*expected_identity = pipes[i].identity = *identity_seq++;
			}
			break;
		}

	return result;
}

static bool
new_last(orafce_pipe *p, void *ptr, size_t size)
{
	queue_item *aux_q;

	if (p->count >= p->limit && p->limit != -1)
		return false;

	if (p->limit == -1 &&
		p->count > 0 &&
		(p->size + size + sizeof(queue_item) > 8 * 1024))
		return false;

	if (p->items == NULL)
	{
		if (NULL == (p->items = ora_salloc(sizeof(queue_item))))
			return false;
		p->items->next_item = NULL;
		p->items->ptr = ptr;
		p->last_item = p->items;
		p->count = 1;
		return true;
	}

	if (NULL == (aux_q = ora_salloc(sizeof(queue_item))))
		return false;

	p->last_item->next_item = aux_q;
	p->last_item = aux_q;

	aux_q->next_item = NULL;
	aux_q->ptr = ptr;

	p->count += 1;

	return true;
}

static void*
remove_first(orafce_pipe *p, bool *found)
{
	struct _queue_item *q;
	void	   *ptr = NULL;

	*found = false;

	if (NULL != (q = p->items))
	{
		p->count -= 1;
		ptr = q->ptr;
		p->items = q->next_item;

		*found = true;

		ora_sfree(q);
		if (p->items == NULL && !p->registered)
		{
			ora_sfree(p->pipe_name);

			if (p->creator)
			{
				ora_sfree(p->creator);
				p->creator = NULL;
			}

			p->is_valid = false;
		}

	}

	return ptr;
}

/* copy message to local memory, if exists */
static message_buffer*
get_from_pipe(text *pipe_name,
			  bool *found,
			  long *identity, bool *identity_alarm)
{
	orafce_pipe *p;
	bool		created;
	message_buffer *result = NULL;

	if (!ora_lock_shmem(SHMEMMSGSZ, MAX_PIPES, MAX_EVENTS, MAX_LOCKS, false))
		return NULL;

	if (NULL != (p = find_pipe(pipe_name, &created, false, identity, identity_alarm)))
	{
		if (!created)
		{
			message_buffer *shm_msg;

			if (NULL != (shm_msg = remove_first(p, found)))
			{
				p->size -= shm_msg->size;

				result = (message_buffer*) MemoryContextAlloc(TopMemoryContext, shm_msg->size);
				memcpy(result, shm_msg, shm_msg->size);
				ora_sfree(shm_msg);
			}
		}
	}

	LWLockRelease(shmem_lockid);

	return result;
}

/*
 * if ptr is null, then only register pipe
 */
static bool
add_to_pipe(text *pipe_name,
			message_buffer *ptr,
			int limit, bool limit_is_valid,
			long *identity, bool *identity_alarm)
{
	bool		created;
	bool		result = false;
	message_buffer *sh_ptr;

	if (!ora_lock_shmem(SHMEMMSGSZ, MAX_PIPES, MAX_EVENTS, MAX_LOCKS,false))
		return false;

	for (;;)
	{
		orafce_pipe *p;

		if (NULL != (p = find_pipe(pipe_name, &created, false, identity, identity_alarm)))
		{
			if (created)
				p->registered = ptr == NULL;

			if (limit_is_valid && (created || (p->limit < limit)))
				p->limit = limit;

			if (ptr != NULL)
			{
				if (NULL != (sh_ptr = ora_salloc(ptr->size)))
				{
					memcpy(sh_ptr,ptr,ptr->size);
					if (new_last(p, sh_ptr, ptr->size))
					{
						p->size += ptr->size;
						result = true;
						break;
					}
					ora_sfree(sh_ptr);
				}
				if (created)
				{
					/* I created new pipe, but haven't memory for new value */
					ora_sfree(p->pipe_name);
					p->is_valid = false;
					result = false;
				}
			}
			else
				result = true;
		}
		break;
	}
	LWLockRelease(shmem_lockid);
	return result;
}

static void
remove_pipe(text *pipe_name, bool purge)
{
	orafce_pipe *p;
	bool		created;

	if (NULL != (p = find_pipe(pipe_name, &created, true, NULL, NULL)))
	{
		queue_item *q = p->items;
		while (q != NULL)
		{
			queue_item *aux_q;

			aux_q = q->next_item;
			if (q->ptr)
				ora_sfree(q->ptr);
			ora_sfree(q);
			q = aux_q;
		}
		p->items = NULL;
		p->size = 0;
		p->count = 0;
		if (!(purge && p->registered))
		{
			ora_sfree(p->pipe_name);
			p->is_valid = false;

			if (p->creator)
			{
				ora_sfree(p->creator);
				p->creator = NULL;
			}
		}
	}
}

Datum
dbms_pipe_next_item_type (PG_FUNCTION_ARGS)
{
	PG_RETURN_INT32(input_buffer != NULL ? input_buffer->next->type : IT_NO_MORE_ITEMS);
}

static void
reset_buffer(message_buffer *buffer, int32 size)
{
	memset(buffer, 0, size);
	buffer->size = message_buffer_size;
	buffer->items_count = 0;
	buffer->next = message_buffer_get_content(buffer);
}

static message_buffer*
check_buffer(message_buffer *buffer, int32 size)
{
	if (buffer == NULL)
	{
		buffer = (message_buffer*) MemoryContextAlloc(TopMemoryContext, size);
		if (buffer == NULL)
			ereport(ERROR,
					(errcode(ERRCODE_OUT_OF_MEMORY),
					 errmsg("out of memory"),
					 errdetail("Failed while allocation block %d bytes in memory.", size)));

		reset_buffer(buffer, size);
	}

	return buffer;
}

Datum
dbms_pipe_pack_message_text(PG_FUNCTION_ARGS)
{
	text	   *str = PG_GETARG_TEXT_PP(0);

	output_buffer = check_buffer(output_buffer, LOCALMSGSZ);
	pack_field(output_buffer, IT_VARCHAR,
		VARSIZE_ANY_EXHDR(str), VARDATA_ANY(str), InvalidOid);

	PG_RETURN_VOID();
}

Datum
dbms_pipe_pack_message_date(PG_FUNCTION_ARGS)
{
	DateADT	dt = PG_GETARG_DATEADT(0);

	output_buffer = check_buffer(output_buffer, LOCALMSGSZ);
	pack_field(output_buffer, IT_DATE,
			   sizeof(dt), &dt, InvalidOid);

	PG_RETURN_VOID();
}

Datum
dbms_pipe_pack_message_timestamp(PG_FUNCTION_ARGS)
{
	TimestampTz dt = PG_GETARG_TIMESTAMPTZ(0);

	output_buffer = check_buffer(output_buffer, LOCALMSGSZ);
	pack_field(output_buffer, IT_TIMESTAMPTZ,
			   sizeof(dt), &dt, InvalidOid);

	PG_RETURN_VOID();
}

Datum
dbms_pipe_pack_message_number(PG_FUNCTION_ARGS)
{
	Numeric	num = PG_GETARG_NUMERIC(0);

	output_buffer = check_buffer(output_buffer, LOCALMSGSZ);
	pack_field(output_buffer, IT_NUMBER,
			   VARSIZE(num) - VARHDRSZ, VARDATA(num), InvalidOid);

	PG_RETURN_VOID();
}

Datum
dbms_pipe_pack_message_bytea(PG_FUNCTION_ARGS)
{
	bytea	   *data = PG_GETARG_BYTEA_P(0);

	output_buffer = check_buffer(output_buffer, LOCALMSGSZ);
	pack_field(output_buffer, IT_BYTEA,
		VARSIZE_ANY_EXHDR(data), VARDATA_ANY(data), InvalidOid);

	PG_RETURN_VOID();
}

static void
init_args_3(FunctionCallInfo info, Datum arg0, Datum arg1, Datum arg2)
{
#if PG_VERSION_NUM >= 120000

	info->args[0].value = arg0;
	info->args[1].value = arg1;
	info->args[2].value = arg2;
	info->args[0].isnull = false;
	info->args[1].isnull = false;
	info->args[2].isnull = false;

#else

	info->arg[0] = arg0;
	info->arg[1] = arg1;
	info->arg[2] = arg2;
	info->argnull[0] = false;
	info->argnull[1] = false;
	info->argnull[2] = false;

#endif
}

/*
 *  We can serialize only typed record
 */
Datum
dbms_pipe_pack_message_record(PG_FUNCTION_ARGS)
{
	HeapTupleHeader rec = PG_GETARG_HEAPTUPLEHEADER(0);
	Oid			tupType;
	bytea	   *data;

#if PG_VERSION_NUM >= 120000

	LOCAL_FCINFO(info, 3);

#else

	FunctionCallInfoData info_data;
	FunctionCallInfo info = &info_data;

#endif

	tupType = HeapTupleHeaderGetTypeId(rec);

	/*
	 * Normally one would call record_send() using DirectFunctionCall3,
	 * but that does not work since record_send wants to cache some data
	 * using fcinfo->flinfo->fn_extra.  So we need to pass it our own
	 * flinfo parameter.
	 */
	InitFunctionCallInfoData(*info, fcinfo->flinfo, 3, InvalidOid, NULL, NULL);
	init_args_3(info, PointerGetDatum(rec), ObjectIdGetDatum(tupType), Int32GetDatum(-1));

	data = (bytea*) DatumGetPointer(record_send(info));

	output_buffer = check_buffer(output_buffer, LOCALMSGSZ);
	pack_field(output_buffer, IT_RECORD,
			   VARSIZE(data), VARDATA(data), tupType);

	PG_RETURN_VOID();
}

static Datum
dbms_pipe_unpack_message(PG_FUNCTION_ARGS, message_data_type dtype)
{
	Oid			tupType;
	void	   *ptr;
	int32		size;
	Datum		result;
	message_data_type next_type;
	message_data_type type;

	if (input_buffer == NULL ||
		input_buffer->items_count <= 0 ||
		input_buffer->next == NULL ||
		input_buffer->next->type == IT_NO_MORE_ITEMS)
		PG_RETURN_NULL();

	next_type = input_buffer->next->type;
	if (next_type != dtype)
		ereport(ERROR,
				(errcode(ERRCODE_DATATYPE_MISMATCH),
				 errmsg("datatype mismatch"),
				 errdetail("unpack unexpected type: %d", next_type)));

	ptr = unpack_field(input_buffer, &type, &size, &tupType);
	Assert(ptr != NULL);

	switch (type)
	{
		case IT_TIMESTAMPTZ:
			result = TimestampTzGetDatum(*(TimestampTz*)ptr);
			break;
		case IT_DATE:
			result = DateADTGetDatum(*(DateADT*)ptr);
			break;
		case IT_VARCHAR:
		case IT_NUMBER:
		case IT_BYTEA:
			result = PointerGetDatum(cstring_to_text_with_len(ptr, size));
			break;
		case IT_RECORD:
		{
#if PG_VERSION_NUM >= 120000

			LOCAL_FCINFO(info, 3);

#else

			FunctionCallInfoData info_data;
			FunctionCallInfo info = &info_data;

#endif

			StringInfoData	buf;
			text		   *data = cstring_to_text_with_len(ptr, size);

			buf.data = VARDATA(data);
			buf.len = VARSIZE(data) - VARHDRSZ;
			buf.maxlen = buf.len;
			buf.cursor = 0;

			/*
			 * Normally one would call record_recv() using DirectFunctionCall3,
			 * but that does not work since record_recv wants to cache some data
			 * using fcinfo->flinfo->fn_extra.  So we need to pass it our own
			 * flinfo parameter.
			 */
			InitFunctionCallInfoData(*info, fcinfo->flinfo, 3, InvalidOid, NULL, NULL);
			init_args_3(info, PointerGetDatum(&buf), ObjectIdGetDatum(tupType), Int32GetDatum(-1));

			result = record_recv(info);
			break;
		}
		default:
			elog(ERROR, "unexpected type: %d", type);
			result = (Datum) 0;	/* keep compiler quiet */
	}

	if (input_buffer->items_count == 0)
	{
		pfree(input_buffer);
		input_buffer = NULL;
	}

	PG_RETURN_DATUM(result);
}

Datum
dbms_pipe_unpack_message_text(PG_FUNCTION_ARGS)
{
	return dbms_pipe_unpack_message(fcinfo, IT_VARCHAR);
}

Datum
dbms_pipe_unpack_message_date(PG_FUNCTION_ARGS)
{
	return dbms_pipe_unpack_message(fcinfo, IT_DATE);
}

Datum
dbms_pipe_unpack_message_timestamp(PG_FUNCTION_ARGS)
{
	return dbms_pipe_unpack_message(fcinfo, IT_TIMESTAMPTZ);
}

Datum
dbms_pipe_unpack_message_number(PG_FUNCTION_ARGS)
{
	return dbms_pipe_unpack_message(fcinfo, IT_NUMBER);
}

Datum
dbms_pipe_unpack_message_bytea(PG_FUNCTION_ARGS)
{
	return dbms_pipe_unpack_message(fcinfo, IT_BYTEA);
}

Datum
dbms_pipe_unpack_message_record(PG_FUNCTION_ARGS)
{
	return dbms_pipe_unpack_message(fcinfo, IT_RECORD);
}


#define WATCH_PRE(t, et, c) \
et = GetNowFloat() + (float8)t; c = 0; (void) c;\
do \
{

#define WATCH_TM_POST(t,et,c) \
if (GetNowFloat() >= et) \
PG_RETURN_INT32(RESULT_TIMEOUT); \
if (cycle++ % 100 == 0) \
CHECK_FOR_INTERRUPTS(); \
pg_usleep(10000L); \
} while(true && t != 0);


Datum
dbms_pipe_receive_message(PG_FUNCTION_ARGS)
{
	text	   *pipe_name = NULL;
	int			timeout;
	bool		found = false;
	instr_time	start_time;
	int32		result = RESULT_TIMEOUT;
	long		identity = NOT_ASSIGNED_IDENTITY;
	bool		identity_alarm;

#if PG_VERSION_NUM < 130000

	long		cycle = 0;

#endif

	if (PG_ARGISNULL(0))
		ereport(ERROR,
				(errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
				 errmsg("pipe name is NULL"),
				 errdetail("Pipename may not be NULL.")));
	else
		pipe_name = PG_GETARG_TEXT_P(0);

	if (!PG_ARGISNULL(1))
	{
		timeout = PG_GETARG_INT32(1);
		if (timeout < 0)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("negative timeout is not allowed")));

		if (timeout > MAXWAIT)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("timeout is too large (maximum: %d)", MAXWAIT)));
	}
	else
		timeout = MAXWAIT;

	if (input_buffer)
	{
		pfree(input_buffer);
		input_buffer = NULL;
	}

	INSTR_TIME_SET_CURRENT(start_time);

	for (;;)
	{
		input_buffer = get_from_pipe(pipe_name, &found, &identity, &identity_alarm);
		if (found)
		{
			if (input_buffer)
				input_buffer->next = message_buffer_get_content(input_buffer);

			result = RESULT_DATA;
			break;
		}

		if (identity_alarm)
			break;

		if (timeout > 0)
		{
			instr_time	cur_time;
			long		cur_timeout;

			INSTR_TIME_SET_CURRENT(cur_time);
			INSTR_TIME_SUBTRACT(cur_time, start_time);

			cur_timeout = timeout * 1000L - (long) INSTR_TIME_GET_MILLISEC(cur_time);
			if (cur_timeout <= 0)
				break;

#if PG_VERSION_NUM >= 130000

			/*
			 * Timeout should be less than INT_MAX, but we set 1 sec as protection
			 * against deadlocks.
			 */
			if (cur_timeout > 1000)
				cur_timeout = 1000;

			if (ConditionVariableTimedSleep(pipe_cv, cur_timeout, PG_WAIT_EXTENSION))
			{
				/* exit on timeout */
				INSTR_TIME_SET_CURRENT(cur_time);
				INSTR_TIME_SUBTRACT(cur_time, start_time);

				cur_timeout = timeout * 1000L - (long) INSTR_TIME_GET_MILLISEC(cur_time);
				if (cur_timeout <= 0)
					break;
			}

#else

			if (cycle++ % 10)
				CHECK_FOR_INTERRUPTS();

			pg_usleep(10000L);

			/* exit on timeout */
			INSTR_TIME_SET_CURRENT(cur_time);
			INSTR_TIME_SUBTRACT(cur_time, start_time);

			cur_timeout = timeout * 1000L - (long) INSTR_TIME_GET_MILLISEC(cur_time);
			if (cur_timeout <= 0)
				break;

#endif

		}
		else
			break;
	}

#if PG_VERSION_NUM >= 130000

	ConditionVariableCancelSleep();

	if (result == RESULT_DATA)
		ConditionVariableBroadcast(pipe_cv);

#endif

	PG_RETURN_INT32(result);
}

Datum
dbms_pipe_send_message(PG_FUNCTION_ARGS)
{
	text	   *pipe_name = NULL;
	int			timeout;
	int			limit = 0;
	bool		valid_limit;
	instr_time	start_time;
	int32		result = RESULT_TIMEOUT;
	long		identity = NOT_ASSIGNED_IDENTITY;
	bool		identity_alarm;

#if PG_VERSION_NUM < 130000

	long		cycle = 0;

#endif

	if (PG_ARGISNULL(0))
		ereport(ERROR,
				(errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
				 errmsg("pipe name is NULL"),
				 errdetail("Pipename may not be NULL.")));
	else
		pipe_name = PG_GETARG_TEXT_P(0);

	output_buffer = check_buffer(output_buffer, LOCALMSGSZ);

	if (!PG_ARGISNULL(1))
	{
		timeout = PG_GETARG_INT32(1);
		if (timeout < 0)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("negative timeout is not allowed")));

		if (timeout > MAXWAIT)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("timeout is too large (maximum: %d)", MAXWAIT)));
	}
	else
		timeout = MAXWAIT;

	if (PG_ARGISNULL(2))
		valid_limit = false;
	else
	{
		limit = PG_GETARG_INT32(2);
		valid_limit = true;
	}

	INSTR_TIME_SET_CURRENT(start_time);

	for (;;)
	{
		if (add_to_pipe(pipe_name, output_buffer,
						limit, valid_limit,
						&identity, &identity_alarm))
		{
			result = RESULT_DATA;
			break;
		}

		if (identity_alarm)
			break;

		if (timeout > 0)
		{
			instr_time	cur_time;
			long		cur_timeout;

			INSTR_TIME_SET_CURRENT(cur_time);
			INSTR_TIME_SUBTRACT(cur_time, start_time);

			cur_timeout = timeout * 1000L - (long) INSTR_TIME_GET_MILLISEC(cur_time);
			if (cur_timeout <= 0)
				break;

#if PG_VERSION_NUM >= 130000

			if (cur_timeout > 1000)
				cur_timeout = 1000;

			if (ConditionVariableTimedSleep(pipe_cv, cur_timeout, PG_WAIT_EXTENSION))
			{
				/* exit on timeout */
				INSTR_TIME_SET_CURRENT(cur_time);
				INSTR_TIME_SUBTRACT(cur_time, start_time);

				cur_timeout = timeout * 1000L - (long) INSTR_TIME_GET_MILLISEC(cur_time);
				if (cur_timeout <= 0)
					break;
			}

#else

			if (cycle++ % 10)
				CHECK_FOR_INTERRUPTS();

			pg_usleep(10000L);

			/* exit on timeout */
			INSTR_TIME_SET_CURRENT(cur_time);
			INSTR_TIME_SUBTRACT(cur_time, start_time);

			cur_timeout = timeout * 1000L - (long) INSTR_TIME_GET_MILLISEC(cur_time);
			if (cur_timeout <= 0)
				break;

#endif

		}
		else
			break;
	}

#if PG_VERSION_NUM >= 130000

	ConditionVariableCancelSleep();

	if (result == RESULT_DATA)
		ConditionVariableBroadcast(pipe_cv);

#endif

	reset_buffer(output_buffer, LOCALMSGSZ);

	PG_RETURN_INT32(result);
}

Datum
dbms_pipe_unique_session_name(PG_FUNCTION_ARGS)
{
	StringInfoData strbuf;
	float8		endtime;
	int			cycle = 0;
	int			timeout = 10;

	WATCH_PRE(timeout, endtime, cycle);
	if (ora_lock_shmem(SHMEMMSGSZ, MAX_PIPES,MAX_EVENTS,MAX_LOCKS,false))
	{
		text	   *result;

		initStringInfo(&strbuf);
		appendStringInfo(&strbuf,"PG$PIPE$%d$%d",sid, MyProcPid);

		result = cstring_to_text_with_len(strbuf.data, strbuf.len);
		pfree(strbuf.data);
		LWLockRelease(shmem_lockid);

		PG_RETURN_TEXT_P(result);
	}
	WATCH_TM_POST(timeout, endtime, cycle);
	LOCK_ERROR();

	PG_RETURN_NULL();
}

#define DB_PIPES_COLS		6

Datum
dbms_pipe_list_pipes(PG_FUNCTION_ARGS)
{
	FuncCallContext *funcctx;
	TupleDesc	tupdesc;
	AttInMetadata *attinmeta;
	PipesFctx  *fctx;
	float8		endtime;
	int			cycle;
	int			timeout = 10;

	if (SRF_IS_FIRSTCALL())
	{
		int		i;
		MemoryContext  oldcontext;
		bool has_lock = false;

		WATCH_PRE(timeout, endtime, cycle);
		if (ora_lock_shmem(SHMEMMSGSZ, MAX_PIPES, MAX_EVENTS, MAX_LOCKS, false))
		{
			has_lock = true;
			break;
		}
		WATCH_TM_POST(timeout, endtime, cycle);
		if (!has_lock)
			LOCK_ERROR();

		funcctx = SRF_FIRSTCALL_INIT();
		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);

		fctx = palloc(sizeof(PipesFctx));
		funcctx->user_fctx = fctx;
		fctx->pipe_nth = 0;

#if PG_VERSION_NUM >= 120000

		tupdesc = CreateTemplateTupleDesc(DB_PIPES_COLS);

#else

		tupdesc = CreateTemplateTupleDesc(DB_PIPES_COLS, false);

#endif

		i = 0;
		TupleDescInitEntry(tupdesc, ++i, "name",    VARCHAROID, -1, 0);
		TupleDescInitEntry(tupdesc, ++i, "items",   INT4OID,    -1, 0);
		TupleDescInitEntry(tupdesc, ++i, "size",    INT4OID,    -1, 0);
		TupleDescInitEntry(tupdesc, ++i, "limit",   INT4OID,    -1, 0);
		TupleDescInitEntry(tupdesc, ++i, "private", BOOLOID,    -1, 0);
		TupleDescInitEntry(tupdesc, ++i, "owner",   VARCHAROID, -1, 0);
		Assert(i == DB_PIPES_COLS);

		attinmeta = TupleDescGetAttInMetadata(tupdesc);
		funcctx->attinmeta = attinmeta;

		MemoryContextSwitchTo(oldcontext);
	}

	funcctx = SRF_PERCALL_SETUP();
	fctx = (PipesFctx *) funcctx->user_fctx;

	while (fctx->pipe_nth < MAX_PIPES)
	{
		if (pipes[fctx->pipe_nth].is_valid)
		{
			Datum		result;
			HeapTuple	tuple;
			char	   *values[DB_PIPES_COLS];
			char		items[16];
			char		size[16];
			char		limit[16];

			/* name */
			values[0] = pipes[fctx->pipe_nth].pipe_name;
			/* items */
			snprintf(items, lengthof(items), "%d", pipes[fctx->pipe_nth].count);
			values[1] = items;
			/* items */
			snprintf(size, lengthof(size), "%d", pipes[fctx->pipe_nth].size);
			values[2] = size;
			/* limit */
			if (pipes[fctx->pipe_nth].limit != -1)
			{
				snprintf(limit, lengthof(limit), "%d", pipes[fctx->pipe_nth].limit);
				values[3] = limit;
			}
			else
				values[3] = NULL;
			/* private */
			values[4] = (pipes[fctx->pipe_nth].creator ? "true" : "false");
			/* owner */
			values[5] = pipes[fctx->pipe_nth].creator;

			tuple = BuildTupleFromCStrings(funcctx->attinmeta, values);
			result = HeapTupleGetDatum(tuple);

			fctx->pipe_nth += 1;
			SRF_RETURN_NEXT(funcctx, result);
		}
		fctx->pipe_nth += 1;
	}

	LWLockRelease(shmem_lockid);
	SRF_RETURN_DONE(funcctx);
}

/*
 * secondary functions
 */

/*
 * Registration explicit pipes
 *   dbms_pipe.create_pipe(pipe_name varchar, limit := -1 int, private := false bool);
 */
Datum
dbms_pipe_create_pipe(PG_FUNCTION_ARGS)
{
	text	   *pipe_name = NULL;
	int			limit = 0;
	bool		is_private;
	bool		limit_is_valid = false;
	bool		created;
	float8		endtime;
	int			cycle;
	int			timeout = 10;

	if (PG_ARGISNULL(0))
		ereport(ERROR,
				(errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
				 errmsg("pipe name is NULL"),
				 errdetail("Pipename may not be NULL.")));
	else
		pipe_name = PG_GETARG_TEXT_P(0);

	if (!PG_ARGISNULL(1))
	{
		limit = PG_GETARG_INT32(1);
		limit_is_valid = true;
	}

	is_private = PG_ARGISNULL(2) ? false : PG_GETARG_BOOL(2);

	WATCH_PRE(timeout, endtime, cycle);
	if (ora_lock_shmem(SHMEMMSGSZ, MAX_PIPES,MAX_EVENTS,MAX_LOCKS,false))
	{
		orafce_pipe *p;
		if (NULL != (p = find_pipe(pipe_name, &created, false, NULL, NULL)))
		{
			if (!created)
			{
				ereport(ERROR,
						(errcode(ERRCODE_DUPLICATE_OBJECT),
						 errmsg("pipe creation error"),
						 errdetail("Pipe is registered.")));
			}
			if (is_private)
			{
				char	   *user;

				p->uid = GetUserId();

				user = (char*)DirectFunctionCall1(namein,
					    CStringGetDatum(GetUserNameFromId(p->uid, false)));

				p->creator = ora_sstrcpy(user);
				pfree(user);
			}
			p->limit = limit_is_valid ? limit : -1;
			p->registered = true;

			LWLockRelease(shmem_lockid);
			PG_RETURN_VOID();
		}
	}
	WATCH_TM_POST(timeout, endtime, cycle);
	LOCK_ERROR();

	PG_RETURN_VOID();
}

/*
 * Clean local input, output buffers
 */
Datum
dbms_pipe_reset_buffer(PG_FUNCTION_ARGS)
{
	if (output_buffer != NULL)
	{
		pfree(output_buffer);
		output_buffer = NULL;
	}

	if (input_buffer != NULL)
	{
		pfree(input_buffer);
		input_buffer = NULL;
	}

	PG_RETURN_VOID();
}

/*
 * Remove all stored messages in pipe. Remove implicit created
 * pipe.
 */
Datum
dbms_pipe_purge(PG_FUNCTION_ARGS)
{
	text	   *pipe_name = PG_GETARG_TEXT_P(0);
	float8		endtime;
	int			cycle = 0;
	int			timeout = 10;

	WATCH_PRE(timeout, endtime, cycle);
	if (ora_lock_shmem(SHMEMMSGSZ, MAX_PIPES,MAX_EVENTS,MAX_LOCKS,false))
	{

		remove_pipe(pipe_name, true);
		LWLockRelease(shmem_lockid);

		PG_RETURN_VOID();
	}
	WATCH_TM_POST(timeout, endtime, cycle);
	LOCK_ERROR();

#if PG_VERSION_NUM >= 130000

	ConditionVariableBroadcast(pipe_cv);

#endif

	PG_RETURN_VOID();
}

/*
 * Remove pipe if exists
 */
Datum
dbms_pipe_remove_pipe(PG_FUNCTION_ARGS)
{
	text	   *pipe_name = PG_GETARG_TEXT_P(0);
	float8		endtime;
	int			cycle = 0;
	int			timeout = 10;

	WATCH_PRE(timeout, endtime, cycle);
	if (ora_lock_shmem(SHMEMMSGSZ, MAX_PIPES,MAX_EVENTS,MAX_LOCKS,false))
	{
		remove_pipe(pipe_name, false);
		LWLockRelease(shmem_lockid);

		PG_RETURN_VOID();
	}
	WATCH_TM_POST(timeout, endtime, cycle);
	LOCK_ERROR();

#if PG_VERSION_NUM >= 130000

	ConditionVariableBroadcast(pipe_cv);

#endif

	PG_RETURN_VOID();
}

/*
 * Some void udf which I can't wrap in sql
 */
Datum
dbms_pipe_create_pipe_2(PG_FUNCTION_ARGS)
{
	Datum	arg1;
	int		limit = -1;

	if (PG_ARGISNULL(0))
		ereport(ERROR,
				(errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
				 errmsg("pipe name is NULL"),
				 errdetail("Pipename may not be NULL.")));

	arg1 = PG_GETARG_DATUM(0);

	if (!PG_ARGISNULL(1))
		limit = PG_GETARG_INT32(1);

	DirectFunctionCall3(dbms_pipe_create_pipe,
						arg1,
						Int32GetDatum(limit),
						BoolGetDatum(false));

	PG_RETURN_VOID();
}

Datum
dbms_pipe_create_pipe_1(PG_FUNCTION_ARGS)
{
	Datum	arg1;

	if (PG_ARGISNULL(0))
		ereport(ERROR,
				(errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
				 errmsg("pipe name is NULL"),
				 errdetail("Pipename may not be NULL.")));

	arg1 = PG_GETARG_DATUM(0);

	DirectFunctionCall3(dbms_pipe_create_pipe,
						arg1,
						(Datum) -1,
						BoolGetDatum(false));

	PG_RETURN_VOID();
}

Datum
dbms_pipe_pack_message_integer(PG_FUNCTION_ARGS)
{
	/* Casting from int4 to numeric */
	DirectFunctionCall1(dbms_pipe_pack_message_number,
				DirectFunctionCall1(int4_numeric, PG_GETARG_DATUM(0)));

	PG_RETURN_VOID();
}

Datum
dbms_pipe_pack_message_bigint(PG_FUNCTION_ARGS)
{
	/* Casting from int8 to numeric */
	DirectFunctionCall1(dbms_pipe_pack_message_number,
				DirectFunctionCall1(int8_numeric, PG_GETARG_DATUM(0)));

	PG_RETURN_VOID();
}