Transitional commit.

Author: danolivo

contrib/pg_exchange/dmq.h

@@ -21,8 +21,10 @@ typedef int8 DmqSenderId;
 
 extern void dmq_init(const char *library_name);
 
-extern DmqDestinationId dmq_destination_add(char *connstr, char *sender_name,
-											char *receiver_name, int ping_period);
+extern DmqDestinationId dmq_destination_add(char *connstr,
+											char *sender_name,
+											char *receiver_name,
+											int ping_period);
 extern DmqConnState dmq_get_destination_status(DmqDestinationId dest_id);
 extern void dmq_destination_drop(const char *receiver_name);
 
@@ -37,11 +39,13 @@ extern const char *dmq_sender_name(DmqSenderId id);
 extern DmqDestinationId dmq_remote_id(const char *name);
 
 extern const char *
-dmq_pop(DmqSenderId *sender_id, void **msg, Size *len, uint64 mask, bool waitMsg);
+dmq_pop(DmqSenderId *sender_id, void **msg, Size *len, uint64 mask,
+		bool waitMsg);
 extern bool dmq_pop_nb(DmqSenderId *sender_id, StringInfo msg, uint64 mask);
 
 extern void dmq_push(DmqDestinationId dest_id, char *stream_name, char *msg);
-extern void dmq_push_buffer(DmqDestinationId dest_id, char *stream_name, const void *buffer, size_t len);
+extern void dmq_push_buffer(DmqDestinationId dest_id, char *stream_name,
+							const void *buffer, size_t len);
 
 typedef void (*dmq_receiver_hook_type) (const char *);
 extern dmq_receiver_hook_type dmq_receiver_start_hook;

contrib/pg_exchange/nodeDistPlanExec.c

@@ -44,6 +44,7 @@ static CustomScanMethods	distplanexec_plan_methods;
 static CustomExecMethods	distplanexec_exec_methods;
 
 char destsName[10] = "DMQ_DESTS";
+char *network_interface;
 
 
 static Node *CreateDistPlanExecState(CustomScan *node);
@@ -257,7 +258,7 @@ EstablishDMQConnections(const lcontext *context, const char *serverName,
 			sprintf(connstr, "host=%s port=%d "
 							 "fallback_application_name=%s",
 							 host, port, senderName);
-
+elog(LOG, "CONN STR: %s", connstr);
 			sub->dest_id = dmq_destination_add(connstr, senderName, receiverName, 10);
 			memcpy(sub->node, receiverName, strlen(receiverName) + 1);
 		}
@@ -391,12 +392,9 @@ ExplainDistPlanExec(CustomScanState *node, List *ancestors, ExplainState *es)
 }
 
 static struct Plan *
-CreateDistExecPlan(PlannerInfo *root,
-					   RelOptInfo *rel,
-					   struct CustomPath *best_path,
-					   List *tlist,
-					   List *clauses,
-					   List *custom_plans)
+CreateDistExecPlan(PlannerInfo *root, RelOptInfo *rel,
+				   struct CustomPath *best_path,
+				   List *tlist, List *clauses, List *custom_plans)
 {
 	CustomScan *distExecNode;
 
@@ -572,7 +570,7 @@ localize_plan(Plan *node, lcontext *context)
 		if (IsExchangePlanNode(node))
 		{
 			List *private = ((CustomScan *) node)->custom_private;
-elog(LOG, "LOCALIZE: exchange");
+
 			if (lnext(lnext(list_head(private))))
 				context->indexinfo = (IndexOptInfo *) lthird(private);
 		}
@@ -582,7 +580,6 @@ elog(LOG, "LOCALIZE: exchange");
 		if (context->foreign_scans != NIL)
 		{
 			CustomScan *css = (CustomScan *) node;
-//			Index scanrelid = ((Scan *) cstmSubPlan1(node))->scanrelid;
 
 			Assert(list_length(context->foreign_scans) == 1);
 			css->custom_plans = list_delete_ptr(css->custom_plans,
@@ -722,10 +719,48 @@ FSExtractServerName(Oid fsid, char **host, int *port)
 	*host = hostname;
 }
 
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <sys/ioctl.h>
+#include <netinet/in.h>
+#include <net/if.h>
+#include <arpa/inet.h>
+#include "common/ip.h"
+
 void
 GetMyServerName(char **host, int *port)
 {
-	*host = pstrdup(LOCALHOST);
+	 int fd;
+	 struct ifreq ifr;
+	 struct addrinfo hintp;
+	 struct addrinfo *result;
+	 char *sipaddr;
+	 struct sockaddr_storage saddr;
+	 int res;
+
+	 fd = socket(AF_INET, SOCK_DGRAM, 0);
+
+	 /* I want to get an IPv4 IP address */
+	 ifr.ifr_addr.sa_family = AF_INET;
+
+	 /* I want IP address attached to "eth0" */
+	 strncpy(ifr.ifr_name, network_interface, IFNAMSIZ-1);
+	 ioctl(fd, SIOCGIFADDR, &ifr);
+	 close(fd);
+
+	 MemSet(&hintp, 0, sizeof(hintp));
+	 hintp.ai_family = AF_INET;
+	 hintp.ai_flags = AI_ALL;
+	 sipaddr = inet_ntoa(((struct sockaddr_in *)&ifr.ifr_addr)->sin_addr);
+	 if ((res = pg_getaddrinfo_all(sipaddr, NULL, &hintp, &result)) != 0)
+		 elog(FATAL, "Cannot resolve network address %s, error=%d.", sipaddr, res);
+	 memcpy(&saddr, result->ai_addr, result->ai_addrlen);
+	 *host = (char *) palloc0(NI_MAXHOST);
+	 if (pg_getnameinfo_all(&saddr, result->ai_addrlen, *host, NI_MAXHOST,
+		 NULL, 0, 0) != 0)
+		 elog(FATAL, "Cannot resolve network name");
+
 	*port = PostPortNumber;
 }
 
@@ -755,8 +790,9 @@ dmq_init_barrier(DMQDestCont *dmq_data, PlanState *child)
 	/* Wait for dmq connection establishing */
 	for (i = 0; i < dmq_data->nservers; i++)
 		while (dmq_get_destination_status(dmq_data->dests[i].dest_id) != Active);
-
+elog(LOG, "DMQ INIT BARRIER");
 	init_exchange_channel(child, (void *) dmq_data);
+	elog(LOG, "END DMQ INIT BARRIER");
 }
 
 static bool
@@ -809,7 +845,7 @@ init_exchange_channel(PlanState *node, void *context)
 		}
 		else
 			state->indexes[i] = j;
-
+elog(LOG, "SendByteMessage: j=%d, dest_id=%d, stream=%s", j, dmq_data->dests[j].dest_id, state->stream);
 		SendByteMessage(dmq_data->dests[j].dest_id, state->stream, ib);
 	}
 	return false;

contrib/pg_exchange/nodeDistPlanExec.h

@@ -27,6 +27,7 @@ typedef struct
 } lcontext;
 
 
+extern char *network_interface;
 extern char destsName[10];
 #define DISTEXECPATHNAME	"DistExecPath"
 

contrib/pg_exchange/pg_exchange.c

@@ -55,13 +55,25 @@ shmem_size(void)
 											 sizeof(DMQDestinations)));
 	return MAXALIGN(size);
 }
+#include "common/ip.h"
+#include "arpa/inet.h"
+#include "sys/socket.h"
+#include <netinet/in.h>
+#include <netdb.h>
+#include <sys/un.h>
+#include "libpq/pqcomm.h"
 
 /*
  * Module load/unload callback
  */
 void
 _PG_init(void)
 {
+	DefineCustomStringVariable("network_interface",
+							   "Set network interface for EXCHANGE communications",
+							   NULL, &network_interface, "lo",
+							   PGC_SUSET, GUC_NOT_IN_SAMPLE, NULL, NULL, NULL);
+
 	EXCHANGE_Init_methods();
 	DUMMYSCAN_Init_methods();
 	EXEC_Hooks_init();
@@ -72,6 +84,47 @@ _PG_init(void)
 
 	old_dmq_receiver_stop_hook = dmq_receiver_stop_hook;
 	dmq_receiver_stop_hook = OnNodeDisconnect;
+/*	{
+		char host[1024];
+		FILE *f = fopen("/home/andrey/PostgresPro/pgcluster/hosts.txt", "rt");
+		struct addrinfo hintp;
+		struct addrinfo *result;
+		MemSet(&hintp, 0, sizeof(hintp));
+//		hintp.ai_socktype = SOCK_STREAM;
+		hintp.ai_family = AF_UNSPEC;
+		hintp.ai_flags = AI_ALL;
+
+		Assert(f != NULL);
+		while (!feof(f))
+		{
+			struct addrinfo *next;
+			int i=0;
+			int res1;
+
+			fscanf(f, "%s", host);
+			res1 = pg_getaddrinfo_all(host, NULL, &hintp, &result);
+			next = result;
+
+			while (next != NULL)
+			{
+				SockAddr a1;
+				int res2;
+				char node[NI_MAXHOST];
+				char service[NI_MAXSERV];
+				char *res;
+
+				res = inet_ntoa(((struct sockaddr_in *)next->ai_addr)->sin_addr);
+				memcpy(&a1.addr, next->ai_addr, next->ai_addrlen);
+				res2 = pg_getnameinfo_all(&a1.addr, next->ai_addrlen, node, NI_MAXHOST,
+						service, NI_MAXSERV, 0);
+				elog(LOG, "[%d] srchost: %s, res: [%d, %d] IP: %s, host: %s, service: %s.",
+						i++, host, res1, res2, res, node, service);
+				next = next->ai_next;
+			}
+			pg_freeaddrinfo_all(hintp.ai_family, result);
+		}
+		fclose(f);
+	} */
 }
 
 Datum

contrib/pg_exchange/stream.c

@@ -325,6 +325,7 @@ RecvTuple(TupleDesc tupdesc, char *streamName, int *status)
 					*status = 3;
 					break;
 				}
+
 				pfree(buf);
 				return (TupleTableSlot *) NULL;
 			}