/*
 * src/test/examples/testlibpqpipeline.c
 *
 *
 * testlibpqpipeline.c
 *		this test program test query pipelining and it's performance impact
 *
 *
 */
#include <stdio.h>
#include <stdlib.h>
#include <sys/time.h>

#include "libpq-fe.h"

// If defined we won't issue more sql commands if the socket's
// write buffer is full
//#define MIN_LOCAL_Q

//#define PRINT_QUERY_PROGRESS

static int testPipelined( PGconn * conn, int totalQueries, int totalQueued, const char * sql );
static int testPipelinedSeries( PGconn * conn, int totalQueries, int totalQueued, int baseline_usecs );


int
testPipelined( PGconn * conn, int totalQueries, int totalQueued, const char * sql )
{
	int nQueriesQueued;
	int nQueriesTotal;
	PGresult * result;
	PGquery * firstQuery;
	PGquery * curQuery;
	
	nQueriesQueued = nQueriesTotal = 0;
	result = NULL;
	firstQuery = curQuery = NULL;
	
	while( nQueriesQueued > 0 || nQueriesTotal < totalQueries ) {
		
		if( PQconsumeInput(conn) == 0 ) {
			printf( "PQconsumeInput ERROR: %s\n", PQerrorMessage(conn) );
			return 1;
		}
		
		do {
			curQuery = PQgetResultQuery(conn);
			
			/* firstQuery is finished */
			if( firstQuery != curQuery )
			{
				//printf( "%p done, curQuery=%p\n", firstQuery, curQuery );
#ifdef PRINT_QUERY_PROGRESS
				printf("-");
#endif
				firstQuery = curQuery;
				nQueriesQueued--;
			}
			
			/* Break if no queries are ready */
			if( !firstQuery || PQisBusy(conn) )
				break;
			
			if( (result = PQgetResult(conn)) != 0 )
				PQclear(result);
		}
		while(1);
		
		if( nQueriesTotal < totalQueries && nQueriesQueued < totalQueued ) {
#ifdef MIN_LOCAL_Q
			int flushResult = PQflush(conn);
			 if( flushResult == -1 ) {
				printf( "PQflush ERROR: %s\n", PQerrorMessage(conn) );
				return 1;
			} else if ( flushResult == 1 )
				continue;
#endif
			if( !PQsendQuery(conn,sql) ){
				printf( "PQsendQuery failed with error: %s\n", PQerrorMessage(conn) );
				return 1;
			}
			if( firstQuery == NULL )
				firstQuery = PQgetSentQuery(conn);
			nQueriesTotal++;
			nQueriesQueued++;
#ifdef PRINT_QUERY_PROGRESS
			printf( "+" );
#endif
		}
	}
#ifdef PRINT_QUERY_PROGRESS
	printf( "\n" );
#endif
	return 0;
}

int testPipelinedSeries( PGconn * conn, int totalQueries, int totalQueued, int baseline_usecs )
{
	int result;
	struct timeval tv1, tv2;
	int secs, usecs;
	
	gettimeofday(&tv1,NULL);
#define TEST_P(q) \
	if( (result = testPipelined(conn,totalQueries,totalQueued,q)) != 0 ) \
		return result;
	TEST_P("INSERT INTO test(id) VALUES (DEFAULT)");
	TEST_P("SELECT * FROM test LIMIT 1");
	TEST_P("SELECT * FROM test");
	TEST_P("DELETE FROM test");
	gettimeofday(&tv2,NULL);
	secs = tv2.tv_sec - tv1.tv_sec;
	usecs = secs * 1000000 + tv2.tv_usec - tv1.tv_usec;
	printf("testPipelinedSeries(%i,%i) took %i.%06i",totalQueries,totalQueued,secs,usecs);
	if (baseline_usecs == 0)
		printf("\n");
	else
		printf(", speedup %.2f\n", (double)baseline_usecs / usecs );
	return usecs;
}

int
main(int argc, char **argv)
{
	PGconn * conn;
	int baseline;
	
	conn = NULL;
	
	/* make a connection to the database */
	conn = PQsetdb(NULL, NULL, NULL, NULL, NULL);

	/* check to see that the backend connection was successfully made */
	if (PQstatus(conn) != CONNECTION_OK)
	{
		fprintf(stderr, "Connection to database failed: %s",
				PQerrorMessage(conn));
		exit(1);
	}

	PQsetPipelining(conn,1);
	PQsetnonblocking(conn,1);
	
	PQclear(PQexec(conn,"CREATE TABLE test ( id SERIAL PRIMARY KEY )"));

	baseline = testPipelinedSeries(conn,10,1,0);
	testPipelinedSeries(conn,10,3,baseline);
	testPipelinedSeries(conn,10,10,baseline);
	
	baseline = testPipelinedSeries(conn,100,1,0);
	testPipelinedSeries(conn,100,3,baseline);
	testPipelinedSeries(conn,100,10,baseline);
	testPipelinedSeries(conn,100,25,baseline);
	testPipelinedSeries(conn,100,50,baseline);
	testPipelinedSeries(conn,100,100,baseline);
	
	PQclear(PQexec(conn,"DROP TABLE test"));
	
	PQfinish(conn);
	
	return 0;
}