commit 03def11a0b8b4a3927b70d6dcc9f0baa3c62d756 Author: mithun Date: Thu Nov 17 16:59:44 2016 +0530 Mithuns patch diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index d04dba7..0a0b820 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&application_name=myapp -postgresql://host1:123,host2:456/somedb +postgresql://host1:123,host2:456/somedb?target_server_type=any&application_name=myapp Components of the hierarchical part of the URI can also be given as parameters. For example: @@ -1382,6 +1382,20 @@ postgresql://%2Fvar%2Flib%2Fpostgresql/dbname + + target_server_type + + + This parameter specifies what kind of server to connect. + Set this to primary, to connect only to primary + server. Set this to any, if you want to connect to + any of the given host. And, the default is any. + A server is recognized as a primary/standby by observering whether it + allows write for this connection. Internally we identify same based on + the output of show transaction_read_only. + + + @@ -7065,6 +7079,16 @@ myEventProc(PGEventId evtId, void *evtInfo, void *passThrough) linkend="libpq-connect-client-encoding"> connection parameter. + + + + + PGTARGETSERVERTYPE + + PGTARGETSERVERTYPE behaves the same as the connection parameter. + + diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c index b4f9ad7..fff5049 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", "PGTARGETSERVERTYPE", 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, "primary") != 0) + { + conn->status = CONNECTION_BAD; + printfPQExpBuffer(&conn->errorMessage, + libpq_gettext("invalid target_server_type value:" + " \"%s\" should be \"any\" or " + "\"primary\"\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_PRIMARY: 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 primary then check if connected + * one is so. + */ + if (conn->target_server_type != NULL && + strcmp(conn->target_server_type, "primary") == 0) + { + conn->status = CONNECTION_OK; + if (!PQsendQuery(conn, + "show transaction_read_only")) + goto error_return; + conn->status = CONNECTION_CHECK_PRIMARY; + 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 primary then check if connected + * one is so. + */ + if (conn->target_server_type != NULL && + strcmp(conn->target_server_type, "primary") == 0) + { + conn->status = CONNECTION_OK; + if (!PQsendQuery(conn, + "show transaction_read_only")) + goto error_return; + conn->status = CONNECTION_CHECK_PRIMARY; + 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_PRIMARY: + { + conn->status = CONNECTION_OK; + if (!PQconsumeInput(conn)) + goto error_return; + + if (PQisBusy(conn)) + { + conn->status = CONNECTION_CHECK_PRIMARY; + 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 (strncmp(val, "on", 2) == 0) + { + 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 \"primary\" " + "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, " @@ -3109,6 +3233,8 @@ freePGconn(PGconn *conn) free(conn->outBuffer); if (conn->rowBuf) free(conn->rowBuf); + if (conn->target_server_type) + free(conn->target_server_type); termPQExpBuffer(&conn->errorMessage); termPQExpBuffer(&conn->workBuffer); @@ -3120,19 +3246,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 +3295,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 +3330,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..f39a110 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_PRIMARY /* Check if it is connected to PRIMARY. */ } ConnStatusType; typedef enum diff --git a/src/interfaces/libpq/libpq-int.h b/src/interfaces/libpq/libpq-int.h index 854ec89..710f3ab 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, primary. */ + /* Optional file to write trace info to */ FILE *Pfdebug;