failover_to_new_master-v1.patch

commit 04c20d33cf5ba6dd244abe21eac03c0bed4db0c8
Author: mithun <mithun@localhost.localdomain>
Date:   Wed Nov 9 20:22:37 2016 +0530

    Failover to new master patch

diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml
index d04dba7..b1ec41b 100644
--- a/doc/src/sgml/libpq.sgml
+++ b/doc/src/sgml/libpq.sgml
@@ -811,7 +811,7 @@ postgresql://localhost/mydb
 postgresql://user@localhost
 postgresql://user:secret@localhost
 postgresql://other@localhost/otherdb?connect_timeout=10&amp;application_name=myapp
-postgresql://host1:123,host2:456/somedb
+postgresql://host1:123,host2:456/somedb?target_server_type=any&amp;application_name=myapp
 </programlisting>
     Components of the hierarchical part of the <acronym>URI</acronym> can also
     be given as parameters.  For example:
@@ -1382,6 +1382,17 @@ postgresql://%2Fvar%2Flib%2Fpostgresql/dbname
        </para>
       </listitem>
      </varlistentry>
+     <varlistentry id="libpq-connect-target-server-type" xreflabel="target_server_type">
+      <term><literal>target_server_type</literal></term>
+      <listitem>
+       <para>
+        This parameter specifies what kind of server to connect.
+        Set this to <literal>master</literal>, to connect only to master
+        server. Set this to <literal>any</literal>, if you want to connect to
+        any of the given host. And, the default is <literal>any</literal>.
+       </para>
+      </listitem>
+    </varlistentry>
     </variablelist>
    </para>
   </sect2>
diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c
index b4f9ad7..7774cab 100644
--- a/src/interfaces/libpq/fe-connect.c
+++ b/src/interfaces/libpq/fe-connect.c
@@ -300,6 +300,10 @@ static const internalPQconninfoOption PQconninfoOptions[] = {
 		"Replication", "D", 5,
 	offsetof(struct pg_conn, replication)},
 
+	{"target_server_type", NULL, NULL, NULL,
+		"Target server type", "", 6,
+	offsetof(struct pg_conn, target_server_type)},
+
 	/* Terminating entry --- MUST BE LAST */
 	{NULL, NULL, NULL, NULL,
 	NULL, NULL, 0}
@@ -336,6 +340,8 @@ static PGconn *makeEmptyPGconn(void);
 static bool fillPGconn(PGconn *conn, PQconninfoOption *connOptions);
 static void freePGconn(PGconn *conn);
 static void closePGconn(PGconn *conn);
+static void release_all_addrinfo(PGconn *conn);
+static void sendTerminateConn(PGconn *conn);
 static PQconninfoOption *conninfo_init(PQExpBuffer errorMessage);
 static PQconninfoOption *parse_connection_string(const char *conninfo,
 						PQExpBuffer errorMessage, bool use_defaults);
@@ -1026,6 +1032,24 @@ connectOptions2(PGconn *conn)
 	}
 
 	/*
+	 * Validate target_server_mode option.
+	 */
+	if (conn->target_server_type)
+	{
+		if (strcmp(conn->target_server_type, "any") != 0
+			&& strcmp(conn->target_server_type, "master") != 0)
+		{
+			conn->status = CONNECTION_BAD;
+			printfPQExpBuffer(&conn->errorMessage,
+							  libpq_gettext("invalid target_server_type value:"
+											" \"%s\" should be \"any\" or "
+											"\"master\"\n"),
+											conn->target_server_type);
+			return false;
+		}
+	}
+
+	/*
 	 * Only if we get this far is it appropriate to try to connect. (We need a
 	 * state flag, rather than just the boolean result of this function, in
 	 * case someone tries to PQreset() the PGconn.)
@@ -1814,6 +1838,7 @@ PQconnectPoll(PGconn *conn)
 			/* Special cases: proceed without waiting. */
 		case CONNECTION_SSL_STARTUP:
 		case CONNECTION_NEEDED:
+		case CONNECTION_CHECK_MASTER:
 			break;
 
 		default:
@@ -2752,27 +2777,6 @@ keep_going:						/* We will come back to here until there is
 					goto error_return;
 				}
 
-				/* We can release the address lists now. */
-				if (conn->connhost != NULL)
-				{
-					int		i;
-
-					for (i = 0; i < conn->nconnhost; ++i)
-					{
-						int		family = AF_UNSPEC;
-
-#ifdef HAVE_UNIX_SOCKETS
-						if (conn->connhost[i].type == CHT_UNIX_SOCKET)
-							family = AF_UNIX;
-#endif
-
-						pg_freeaddrinfo_all(family,
-											conn->connhost[i].addrlist);
-						conn->connhost[i].addrlist = NULL;
-					}
-				}
-				conn->addr_cur = NULL;
-
 				/* Fire up post-connection housekeeping if needed */
 				if (PG_PROTOCOL_MAJOR(conn->pversion) < 3)
 				{
@@ -2782,7 +2786,25 @@ keep_going:						/* We will come back to here until there is
 					return PGRES_POLLING_WRITING;
 				}
 
-				/* Otherwise, we are open for business! */
+				/*
+				 * If asked to connect to a master then check if connected
+				 * one is a master.
+				 */
+				if (conn->target_server_type != NULL &&
+					strcmp(conn->target_server_type, "master") == 0)
+				{
+					conn->status = CONNECTION_OK;
+					if (!PQsendQuery(conn,
+									 "SELECT pg_catalog.pg_is_in_recovery()"))
+						goto error_return;
+					conn->status = CONNECTION_CHECK_MASTER;
+					return PGRES_POLLING_READING;
+				}
+
+				/* We can release the address lists now. */
+				release_all_addrinfo(conn);
+
+				/* We are open for business! */
 				conn->status = CONNECTION_OK;
 				return PGRES_POLLING_OK;
 			}
@@ -2814,10 +2836,112 @@ keep_going:						/* We will come back to here until there is
 					goto error_return;
 			}
 
+			/*
+			 * If asked to connect to a master then check if connected
+			 * one is so.
+			 */
+			if (conn->target_server_type != NULL &&
+				strcmp(conn->target_server_type, "master") == 0)
+			{
+				conn->status = CONNECTION_OK;
+				if (!PQsendQuery(conn,
+								 "SELECT pg_catalog.pg_is_in_recovery()"))
+					goto error_return;
+				conn->status = CONNECTION_CHECK_MASTER;
+				return PGRES_POLLING_READING;
+			}
+
+			/* We can release the address lists now. */
+			release_all_addrinfo(conn);
+
 			/* We are open for business! */
 			conn->status = CONNECTION_OK;
 			return PGRES_POLLING_OK;
 
+		case CONNECTION_CHECK_MASTER:
+			{
+				conn->status = CONNECTION_OK;
+				if (!PQconsumeInput(conn))
+					goto error_return;
+
+				if (PQisBusy(conn))
+				{
+					conn->status = CONNECTION_CHECK_MASTER;
+					return PGRES_POLLING_READING;
+				}
+
+				res = PQgetResult(conn);
+				if (res && (PQresultStatus(res) == PGRES_TUPLES_OK) &&
+					PQntuples(res) == 1)
+				{
+					char *val;
+					val = PQgetvalue(res, 0 , 0);
+					if (val[0] == 't')
+					{
+						PQclear(res);
+
+						/*
+						 * This is a hot standby server terminate this connection
+						 * and try next address.
+						 */
+						appendPQExpBuffer(&conn->errorMessage,
+							libpq_gettext("Connection failed, "
+										  "%s:%s is a hot standby.\n"),
+										  conn->connhost[conn->whichhost].host,
+										  conn->connhost[conn->whichhost].port);
+						conn->status = CONNECTION_OK;
+						sendTerminateConn(conn);
+						pqDropConnection(conn, true);
+
+						if (conn->addr_cur->ai_next != NULL ||
+							conn->whichhost + 1 < conn->nconnhost)
+						{
+							conn->addr_cur = conn->addr_cur->ai_next;
+							conn->status = CONNECTION_NEEDED;
+							goto keep_going;
+						}
+
+						/* No more addresses to try. So we fail. */
+						goto error_return;
+					}
+					PQclear(res);
+
+					/* We can release the address lists now. */
+					release_all_addrinfo(conn);
+
+					/* We are open for business! */
+					conn->status = CONNECTION_OK;
+						return PGRES_POLLING_OK;
+				}
+
+				/*
+				 * Something went wrong with
+				 * "SELECT pg_catalog.pg_is_in_recovery()". We should try next
+				 * addresses.
+				 */
+				if (res)
+					PQclear(res);
+				appendPQExpBuffer(&conn->errorMessage,
+					libpq_gettext("test to check if, %s:%s is a \"master\" "
+								  "failed \n"),
+								  conn->connhost[conn->whichhost].host,
+								  conn->connhost[conn->whichhost].port);
+				conn->status = CONNECTION_OK;
+				sendTerminateConn(conn);
+				pqDropConnection(conn, true);
+
+				if (conn->addr_cur->ai_next != NULL ||
+					conn->whichhost + 1 < conn->nconnhost)
+				{
+					conn->addr_cur = conn->addr_cur->ai_next;
+					conn->status = CONNECTION_NEEDED;
+					goto keep_going;
+				}
+
+				/* No more addresses to try. So we fail. */
+				goto error_return;
+			}
+
 		default:
 			appendPQExpBuffer(&conn->errorMessage,
 							  libpq_gettext("invalid connection state %d, "
@@ -3120,19 +3244,41 @@ freePGconn(PGconn *conn)
 }
 
 /*
- * closePGconn
- *	 - properly close a connection to the backend
- *
- * This should reset or release all transient state, but NOT the connection
- * parameters.  On exit, the PGconn should be in condition to start a fresh
- * connection with the same parameters (see PQreset()).
+ * release_all_addrinfo
+ *   - free addrinfo of all hostconn elements.
  */
+
 static void
-closePGconn(PGconn *conn)
+release_all_addrinfo(PGconn *conn)
 {
-	PGnotify   *notify;
-	pgParameterStatus *pstatus;
+	if (conn->connhost != NULL)
+	{
+		int		i;
+
+		for (i = 0; i < conn->nconnhost; ++i)
+		{
+			int		family = AF_UNSPEC;
+
+#ifdef HAVE_UNIX_SOCKETS
+			if (conn->connhost[i].type == CHT_UNIX_SOCKET)
+				family = AF_UNIX;
+#endif
+
+			pg_freeaddrinfo_all(family,
+								conn->connhost[i].addrlist);
+			conn->connhost[i].addrlist = NULL;
+		}
+	}
+	conn->addr_cur = NULL;
+}
 
+/*
+ * sendTerminateConn
+ * 	 - Send a terminate message to backend.
+ */
+static void
+sendTerminateConn(PGconn *conn)
+{
 	/*
 	 * Note that the protocol doesn't allow us to send Terminate messages
 	 * during the startup phase.
@@ -3147,6 +3293,23 @@ closePGconn(PGconn *conn)
 		pqPutMsgEnd(conn);
 		(void) pqFlush(conn);
 	}
+}
+
+/*
+ * closePGconn
+ *	 - properly close a connection to the backend
+ *
+ * This should reset or release all transient state, but NOT the connection
+ * parameters.  On exit, the PGconn should be in condition to start a fresh
+ * connection with the same parameters (see PQreset()).
+ */
+static void
+closePGconn(PGconn *conn)
+{
+	PGnotify   *notify;
+	pgParameterStatus *pstatus;
+
+	sendTerminateConn(conn);
 
 	/*
 	 * Must reset the blocking status so a possible reconnect will work.
@@ -3165,25 +3328,8 @@ closePGconn(PGconn *conn)
 	conn->asyncStatus = PGASYNC_IDLE;
 	pqClearAsyncResult(conn);	/* deallocate result */
 	resetPQExpBuffer(&conn->errorMessage);
-	if (conn->connhost != NULL)
-	{
-		int		i;
-
-		for (i = 0; i < conn->nconnhost; ++i)
-		{
-			int		family = AF_UNSPEC;
+	release_all_addrinfo(conn);
 
-#ifdef HAVE_UNIX_SOCKETS
-			if (conn->connhost[i].type == CHT_UNIX_SOCKET)
-				family = AF_UNIX;
-#endif
-
-			pg_freeaddrinfo_all(family,
-								conn->connhost[i].addrlist);
-			conn->connhost[i].addrlist = NULL;
-		}
-	}
-	conn->addr_cur = NULL;
 	notify = conn->notifyHead;
 	while (notify != NULL)
 	{
diff --git a/src/interfaces/libpq/libpq-fe.h b/src/interfaces/libpq/libpq-fe.h
index 9ca0756..5e01b57 100644
--- a/src/interfaces/libpq/libpq-fe.h
+++ b/src/interfaces/libpq/libpq-fe.h
@@ -62,7 +62,8 @@ typedef enum
 								 * backend startup. */
 	CONNECTION_SETENV,			/* Negotiating environment. */
 	CONNECTION_SSL_STARTUP,		/* Negotiating SSL. */
-	CONNECTION_NEEDED			/* Internal state: connect() needed */
+	CONNECTION_NEEDED,			/* Internal state: connect() needed */
+	CONNECTION_CHECK_MASTER		/* Check if it is connected to MASTER. */
 } ConnStatusType;
 
 typedef enum
diff --git a/src/interfaces/libpq/libpq-int.h b/src/interfaces/libpq/libpq-int.h
index 854ec89..1ea6943 100644
--- a/src/interfaces/libpq/libpq-int.h
+++ b/src/interfaces/libpq/libpq-int.h
@@ -361,6 +361,9 @@ struct pg_conn
 	char	   *krbsrvname;		/* Kerberos service name */
 #endif
 
+	char	   *target_server_type;	/* Type of server to connect.
+									 * Possible values any, master. */
+
 	/* Optional file to write trace info to */
 	FILE	   *Pfdebug;