PGconn * pgpool_getfreeconn(void)
{
	assert(initialized);
	int id = last_used;
	while (1) {
		id = (id+1) % poolsize;
		PGconn *c = pool[id];
		assert(c);
		PGresult *res;
		int rc;
		// can we get state AND is the server 
		// not blocking and any results?
		while ((rc=PQconsumeInput(c)) && !PQisBusy(c) 
				&& (res=PQgetResult(c))) {
			// we have results. we need to clear those resultsets
			if (reshandler(c, res)) { 
				return NULL;
			}
		}
		if (!rc) {
			log_fatal("pgpool", "Error in PQconsumeInput. %s",
					PQerrorMessage(c));
			return NULL;
		} else if (!PQisBusy(c)) {
			last_used = id;
			return c;
		}
		if (id == last_used) {
			usleep(150000);
		}
	}
}

static void
evpg_query_finished(int sock, short which, void **data)
{
    struct evpg_db_node *dbnode;
    struct evpg_cfg *config;
    const char *querystr;
    void (*cb)(PGresult *, void *); 
    void *usrdata;
    struct event *event;

    config   = data[0];
    querystr = data[1];
    cb       = data[2];
    usrdata  = data[3];
    dbnode   = data[4];
    event    = data[5];

    PQconsumeInput(dbnode->dbconn);

    if (PQisBusy(dbnode->dbconn) == 0)
    {
	PGresult *result;
	result = PQgetResult(dbnode->dbconn);
	cb(result, usrdata);
	PQclear(result);
	free(event);
	free(data);
	evpg_set_ready(config, dbnode);
	return;
    }

    /* this query has not finished */
    event_set(event, sock, EV_READ, (void *)evpg_query_finished, data);
    event_add(event, 0);
}

PGresult * do_postgres_cCommand_execute_async(VALUE self, VALUE connection, PGconn *db, VALUE query) {
  PGresult *response;
  char* str = StringValuePtr(query);

  while ((response = PQgetResult(db))) {
    PQclear(response);
  }

  struct timeval start;
  int retval;

  gettimeofday(&start, NULL);
  retval = PQsendQuery(db, str);

  if (!retval) {
    if (PQstatus(db) != CONNECTION_OK) {
      PQreset(db);

      if (PQstatus(db) == CONNECTION_OK) {
        retval = PQsendQuery(db, str);
      }
      else {
        do_postgres_full_connect(connection, db);
        retval = PQsendQuery(db, str);
      }
    }

    if (!retval) {
      rb_raise(eDO_ConnectionError, "%s", PQerrorMessage(db));
    }
  }

  int socket_fd = PQsocket(db);
  fd_set rset;

  while (1) {
    FD_ZERO(&rset);
    FD_SET(socket_fd, &rset);
    retval = rb_thread_select(socket_fd + 1, &rset, NULL, NULL, NULL);

    if (retval < 0) {
      rb_sys_fail(0);
    }

    if (retval == 0) {
      continue;
    }

    if (PQconsumeInput(db) == 0) {
      rb_raise(eDO_ConnectionError, "%s", PQerrorMessage(db));
    }

    if (PQisBusy(db) == 0) {
      break;
    }
  }

  data_objects_debug(connection, query, &start);
  return PQgetResult(db);
}

static int
esql_postgresql_io(Esql *e)
{
   if (PQstatus(e->backend.db) == CONNECTION_BAD)
     {
        ERR("%s", esql_postgresql_error_get(e));
        return ECORE_FD_ERROR;
     }
   if (e->current == ESQL_CONNECT_TYPE_INIT)
     {
        switch (PQconnectPoll(e->backend.db))
          {
           case PGRES_POLLING_OK:
             return 0;
           case PGRES_POLLING_READING:
             return ECORE_FD_READ;
           case PGRES_POLLING_WRITING:
             return ECORE_FD_WRITE;
           default:
             ERR("%s", esql_postgresql_error_get(e));
             return ECORE_FD_ERROR;
          }
     }
   if (!PQconsumeInput(e->backend.db))
     {
        ERR("%s", esql_postgresql_error_get(e));
        return ECORE_FD_ERROR;
     }
   if (!PQisBusy(e->backend.db)) return 0;
   return ECORE_FD_READ | ECORE_FD_WRITE; /* psql does not provide a method to get read/write mode :( */
}

static void
write_queue(ParallelWriter *self, const void *buffer, uint32 len)
{
	struct iovec	iov[2];

	AssertArg(self->conn != NULL);
	AssertArg(self->queue != NULL);
	AssertArg(len == 0 || buffer != NULL);

	iov[0].iov_base = &len;
	iov[0].iov_len = sizeof(len);
	iov[1].iov_base = (void *) buffer;
	iov[1].iov_len = len;

	for (;;)
	{
		if (QueueWrite(self->queue, iov, 2, DEFAULT_TIMEOUT_MSEC, false))
			return;

		PQconsumeInput(self->conn);
		if (!PQisBusy(self->conn))
		{
			ereport(ERROR,
				(errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
				 errmsg("unexpected reader termination"),
				 errdetail("%s", finish_and_get_message(self))));
		}

		/* retry */
	}
}

/*
 * Send a query and wait for the results by using the asynchronous libpq
 * functions and the backend version of select().
 *
 * We must not use the regular blocking libpq functions like PQexec()
 * since they are uninterruptible by signals on some platforms, such as
 * Windows.
 *
 * We must also not use vanilla select() here since it cannot handle the
 * signal emulation layer on Windows.
 *
 * The function is modeled on PQexec() in libpq, but only implements
 * those parts that are in use in the walreceiver.
 *
 * Queries are always executed on the connection in streamConn.
 */
static PGresult *
libpqrcv_PQexec(const char *query)
{
	PGresult   *result = NULL;
	PGresult   *lastResult = NULL;

	/*
	 * PQexec() silently discards any prior query results on the connection.
	 * This is not required for walreceiver since it's expected that walsender
	 * won't generate any such junk results.
	 */

	/*
	 * Submit a query. Since we don't use non-blocking mode, this also can
	 * block. But its risk is relatively small, so we ignore that for now.
	 */
	if (!PQsendQuery(streamConn, query))
		return NULL;

	for (;;)
	{
		/*
		 * Receive data until PQgetResult is ready to get the result without
		 * blocking.
		 */
		while (PQisBusy(streamConn))
		{
			/*
			 * We don't need to break down the sleep into smaller increments,
			 * and check for interrupts after each nap, since we can just
			 * elog(FATAL) within SIGTERM signal handler if the signal arrives
			 * in the middle of establishment of replication connection.
			 */
			if (!libpq_select(-1))
				continue;		/* interrupted */
			if (PQconsumeInput(streamConn) == 0)
				return NULL;	/* trouble */
		}

		/*
		 * Emulate the PQexec()'s behavior of returning the last result when
		 * there are many. Since walsender will never generate multiple
		 * results, we skip the concatenation of error messages.
		 */
		result = PQgetResult(streamConn);
		if (result == NULL)
			break;				/* query is complete */

		PQclear(lastResult);
		lastResult = result;

		if (PQresultStatus(lastResult) == PGRES_COPY_IN ||
			PQresultStatus(lastResult) == PGRES_COPY_OUT ||
			PQresultStatus(lastResult) == PGRES_COPY_BOTH ||
			PQstatus(streamConn) == CONNECTION_BAD)
			break;
	}

	return lastResult;
}

/*
 * wait until current query finishes ignoring any results, this could be an async command
 * or a cancelation of a query
 * return 1 if Ok; 0 if any error ocurred; -1 if timeout reached
 */
int
wait_connection_availability(PGconn *conn, int timeout)
{
	PGresult   *res;

	while(timeout-- >= 0)
	{
		if (PQconsumeInput(conn) == 0)
		{
			log_warning(_("PQconsumeInput: Query could not be sent to primary. %s\n"),
			            PQerrorMessage(conn));
			return 0;
		}

		if (PQisBusy(conn) == 0)
		{
			res = PQgetResult(conn);
			if (res == NULL)
				break;
			PQclear(res);
		}
		sleep(1);
	}
	if (timeout >= 0)
		return 1;
	else
		return -1;
}

/* Checks whether new data has arrived from the server (on either the snapshot
 * connection or the replication connection, as appropriate). If yes, it is
 * processed, and context->status is set to 1. If no data is available, this
 * function does not block, but returns immediately, and context->status is set
 * to 0. If the data stream has ended, context->status is set to -1. */
int db_client_poll(client_context_t context) {
    int err = 0;

    if (context->sql_conn) {
        /* To make PQgetResult() non-blocking, check PQisBusy() first */
        if (PQisBusy(context->sql_conn)) {
            context->status = 0;
            return err;
        }

        check(err, snapshot_poll(context));
        context->status = 1;

        /* If the snapshot is finished, switch over to the replication stream */
        if (!context->sql_conn) {
            checkRepl(err, context, replication_stream_start(&context->repl));
        }
        return err;

    } else {
        checkRepl(err, context, replication_stream_poll(&context->repl));
        context->status = context->repl.status;
        return err;
    }
}

static void pq_event(evutil_socket_t fd, short event, void *arg) {
  struct connection_struct* database = (struct connection_struct*) arg;
  if (database->queries) {
    if (database->queries->sent == 0) {
      PQsendQuery(database->conn, database->queries->query);
      database->queries->sent = 1;
    }
    if (PQconsumeInput(database->conn) && !PQisBusy(database->conn)) {
      PGresult* res = PQgetResult(database->conn);
      while (res) {
        if (database->queries->callback)
          database->queries->callback(res, database->queries->context, database->queries->query);
        if (database->report_errors && PQresultStatus(res) != PGRES_COMMAND_OK)
          fprintf(stderr, "Query: '%s' returned error\n\t%s\n", database->queries->query, PQresultErrorMessage(res));
        PQclear(res);
        res = PQgetResult(database->conn);
      }
      database->query_count--;
      struct query_struct* old = database->queries;
      database->queries = database->queries->next;
      free(old->query);
      free(old);
      pq_event(fd, event, arg);
    }
  }
}

/*
 * Helper function that actually kicks off the command on the libpq connection.
 */
static void
dispatchCommand(CdbDispatchResult * dispatchResult,
				const char *query_text,
				int query_text_len)
{
	SegmentDatabaseDescriptor *segdbDesc = dispatchResult->segdbDesc;
	TimestampTz beforeSend = 0;
	long secs;
	int	usecs;

	if (DEBUG1 >= log_min_messages)
		beforeSend = GetCurrentTimestamp();

	if (PQisBusy(segdbDesc->conn))
		elog(LOG, "Trying to send to busy connection %s: asyncStatus %d",
				  segdbDesc->whoami,
				  segdbDesc->conn->asyncStatus);

	if (cdbconn_isBadConnection(segdbDesc))
	{
		char *msg = PQerrorMessage(dispatchResult->segdbDesc->conn);
		dispatchResult->stillRunning = false;
		ereport(ERROR,
				(errcode(ERRCODE_GP_INTERCONNECTION_ERROR),
				 errmsg("Connection lost before dispatch to segment %s: %s",
						 dispatchResult->segdbDesc->whoami, msg ? msg : "unknown error")));
	}

	/*
	 * Submit the command asynchronously.
	 */
	if (PQsendGpQuery_shared(dispatchResult->segdbDesc->conn, (char *) query_text, query_text_len) == 0)
	{
		char *msg = PQerrorMessage(dispatchResult->segdbDesc->conn);
		dispatchResult->stillRunning = false;
		ereport(ERROR,
				(errcode(ERRCODE_GP_INTERCONNECTION_ERROR),
				 errmsg("Command could not be dispatch to segment %s: %s",
						 dispatchResult->segdbDesc->whoami, msg ? msg : "unknown error")));
	}

	if (DEBUG1 >= log_min_messages)
	{
		TimestampDifference(beforeSend, GetCurrentTimestamp(), &secs, &usecs);

		if (secs != 0 || usecs > 1000)	/* Time > 1ms? */
			elog(LOG, "time for PQsendGpQuery_shared %ld.%06d", secs, usecs);
	}

	/*
	 * We'll keep monitoring this QE -- whether or not the command
	 * was dispatched -- in order to check for a lost connection
	 * or any other errors that libpq might have in store for us.
	 */
	dispatchResult->stillRunning = true;
	dispatchResult->hasDispatched = true;

	ELOG_DISPATCHER_DEBUG("Command dispatched to QE (%s)", dispatchResult->segdbDesc->whoami);
}

int
pgut_wait(int num, PGconn *connections[], struct timeval *timeout)
{
	/* all connections are busy. wait for finish */
	while (!interrupted)
	{
		int		i;
		fd_set	mask;
		int		maxsock;

		FD_ZERO(&mask);

		maxsock = -1;
		for (i = 0; i < num; i++)
		{
			int	sock;

			if (connections[i] == NULL)
				continue;
			sock = PQsocket(connections[i]);
			if (sock >= 0)
			{
				FD_SET(sock, &mask);
				if (maxsock < sock)
					maxsock = sock;
			}
		}

		if (maxsock == -1)
		{
			errno = ENOENT;
			return -1;
		}

		i = wait_for_sockets(maxsock + 1, &mask, timeout);
		if (i == 0)
			break;	/* timeout */

		for (i = 0; i < num; i++)
		{
			if (connections[i] && FD_ISSET(PQsocket(connections[i]), &mask))
			{
				PQconsumeInput(connections[i]);
				if (PQisBusy(connections[i]))
					continue;
				return i;
			}
		}
	}

	errno = EINTR;
	return -1;
}

int db_postgres_async_resume(db_con_t *_h, int fd, db_res_t **_r, void *_priv)
{
	struct pool_con *con = (struct pool_con *)_priv;
	PGresult *res = NULL;

#ifdef EXTRA_DEBUG
	if (!db_match_async_con(fd, _h)) {
		LM_BUG("no conn match for fd %d", fd);
		abort();
	}
#endif


	db_switch_to_async(_h, con);

	if( PQconsumeInput(CON_CONNECTION(_h)) == 0) {
		LM_ERR("Unable to consume input\n");
		db_switch_to_sync(_h);
		db_store_async_con(_h, con);
		return -1;
	}

	if(PQisBusy(CON_CONNECTION(_h))) {
		async_status = ASYNC_CONTINUE;

		db_switch_to_sync(_h);
		return 1;
	}

	while (1) {
		if ((res = PQgetResult(CON_CONNECTION(_h)))) {
			CON_RESULT(_h) = res;
		} else {
			break;
		}
	}

	if (_r) {
		if (db_postgres_store_result(_h, _r) != 0) {
			LM_ERR("failed to store result\n");
			db_switch_to_sync(_h);
			db_store_async_con(_h, con);
			return -2;
		}
	}

	db_switch_to_sync(_h);
	db_store_async_con(_h, con);

	return 0;
}

static void
close_connections()
{
	if (primary_conn != NULL && PQisBusy(primary_conn) == 1)
		cancel_query(primary_conn, local_options.master_response_timeout);

	if (my_local_conn != NULL)
		PQfinish(my_local_conn);

	if (primary_conn != NULL && primary_conn != my_local_conn)
		PQfinish(primary_conn);

	primary_conn = NULL;
	my_local_conn = NULL;
}

/*
 * call-seq:
 *    conn.fetch()                   -> result or nil
 *    conn.fetch() { |result| ... }  -> obj
 *
 * Fetches the results of the previous Pg::Conn#send call.
 * See there for an example.
 *
 * The result will be +nil+ if there are no more results.
 */
VALUE
pgconn_fetch( VALUE self)
{
    struct pgconn_data *c;
    PGresult *result;
    VALUE res;

    Data_Get_Struct( self, struct pgconn_data, c);
    pg_check_conninvalid( c);
    if (PQconsumeInput( c->conn) == 0)
        pg_raise_connexec( c);
    if (PQisBusy( c->conn) > 0)
        return Qnil;
    result = PQgetResult( c->conn);
    return result == NULL ? Qnil :
        yield_or_return_result( pgresult_new( result, c, Qnil, Qnil));
}

ngx_int_t
ngx_postgres_upstream_get_ack(ngx_http_request_t *r, ngx_connection_t *pgxc,
    ngx_postgres_upstream_peer_data_t *pgdt)
{
    PGresult  *res;

    dd("entering");

    if (!PQconsumeInput(pgdt->pgconn)) {
        dd("returning NGX_ERROR");
        return NGX_ERROR;
    }

    if (PQisBusy(pgdt->pgconn)) {
        dd("returning NGX_AGAIN");
        return NGX_AGAIN;
    }

    /* remove result timeout */
    if (pgxc->read->timer_set) {
        ngx_del_timer(pgxc->read);
    }

    dd("receiving ACK (ready for next query)");

    res = PQgetResult(pgdt->pgconn);
    if (res != NULL) {
        dd("receiving ACK failed");
        ngx_log_error(NGX_LOG_ERR, pgxc->log, 0,
                      "receiving ACK failed: multiple queries(?)");

        PQclear(res);

        dd("returning NGX_HTTP_INTERNAL_SERVER_ERROR");
        return NGX_HTTP_INTERNAL_SERVER_ERROR;
    }

    dd("ACK received successfully");

    pgxc->log->action = "being idle on PostgreSQL database";
    pgdt->state = state_db_idle;

    dd("returning");
    return ngx_postgres_upstream_done(r, r->upstream, pgdt);
}

    bool Connection::ping()
    {
      log_debug("ping()");

      if (PQsendQuery(conn, "select 1") == 0)
      {
        log_debug("failed to send statement \"select 1\" to database in Connection::ping()");
        return false;
      }

      while (true)
      {
        struct pollfd fd;
        fd.fd = PQsocket(conn);
        fd.events = POLLIN;
        log_debug("wait for input on fd " << fd.fd);
        if (::poll(&fd, 1, 10000) != 1)
        {
          log_debug("no data received in Connection::ping()");
          return false;
        }

        log_debug("consumeInput");
        if (PQconsumeInput(conn) == 0)
        {
          log_debug("PQconsumeInput failed in Connection::ping()");
          return false;
        }

        log_debug("check PQisBusy");
        while (PQisBusy(conn) == 0)
        {
          log_debug("PQgetResult");
          PGresult* result = PQgetResult(conn);

          log_debug("PQgetResult => " << static_cast<void*>(result));
          if (result == 0)
            return true;

          log_debug("PQfree");
          PQclear(result);
        }
      }
    }

void
ngx_postgres_keepalive_close_handler(ngx_event_t *ev)
{
    ngx_postgres_upstream_srv_conf_t  *pgscf;
    ngx_postgres_keepalive_cache_t    *item;
    ngx_connection_t                  *c;
    PGresult                          *res;

    dd("entering");

    c = ev->data;
    item = c->data;

    if (c->close) {
        goto close;
    }

    if (PQconsumeInput(item->pgconn) && !PQisBusy(item->pgconn)) {
        res = PQgetResult(item->pgconn);
        if (res == NULL) {
            dd("returning");
            return;
        }

        PQclear(res);

        dd("received result on idle keepalive connection");
        ngx_log_error(NGX_LOG_ERR, c->log, 0,
                      "postgres: received result on idle keepalive connection");
    }

close:

    pgscf = item->srv_conf;

    ngx_postgres_upstream_free_connection(ev->log, c, item->pgconn, pgscf);

    ngx_queue_remove(&item->queue);
    ngx_queue_insert_head(&pgscf->free, &item->queue);

    dd("returning");
}

/* MultiClientResultStatus checks result status for an asynchronous query. */
ResultStatus
MultiClientResultStatus(int32 connectionId)
{
	PGconn *connection = NULL;
	int consumed = 0;
	ConnStatusType connStatusType = CONNECTION_OK;
	ResultStatus resultStatus = CLIENT_INVALID_RESULT_STATUS;

	Assert(connectionId != INVALID_CONNECTION_ID);
	connection = ClientConnectionArray[connectionId];
	Assert(connection != NULL);

	connStatusType = PQstatus(connection);
	if (connStatusType == CONNECTION_BAD)
	{
		ereport(WARNING, (errmsg("could not maintain connection to worker node")));
		return CLIENT_RESULT_UNAVAILABLE;
	}

	/* consume input to allow status change */
	consumed = PQconsumeInput(connection);
	if (consumed != 0)
	{
		int connectionBusy = PQisBusy(connection);
		if (connectionBusy == 0)
		{
			resultStatus = CLIENT_RESULT_READY;
		}
		else
		{
			resultStatus = CLIENT_RESULT_BUSY;
		}
	}
	else
	{
		ereport(WARNING, (errmsg("could not consume data from worker node")));
		resultStatus = CLIENT_RESULT_UNAVAILABLE;
	}

	return resultStatus;
}

/*
 * Wait for the result from a prior asynchronous execution function call.
 *
 * This function offers quick responsiveness by checking for any interruptions.
 *
 * This function emulates the PQexec()'s behavior of returning the last result
 * when there are many.
 *
 * Caller is responsible for the error handling on the result.
 */
PGresult *
pgfdw_get_result(PGconn *conn, const char *query)
{
	PGresult   *last_res = NULL;

	for (;;)
	{
		PGresult   *res;

		while (PQisBusy(conn))
		{
			int			wc;

			/* Sleep until there's something to do */
			wc = WaitLatchOrSocket(MyLatch,
								   WL_LATCH_SET | WL_SOCKET_READABLE,
								   PQsocket(conn),
								   -1L, PG_WAIT_EXTENSION);
			ResetLatch(MyLatch);

			CHECK_FOR_INTERRUPTS();

			/* Data available in socket */
			if (wc & WL_SOCKET_READABLE)
			{
				if (!PQconsumeInput(conn))
					pgfdw_report_error(ERROR, NULL, conn, false, query);
			}
		}

		res = PQgetResult(conn);
		if (res == NULL)
			break;				/* query is complete */

		PQclear(last_res);
		last_res = res;
	}

	return last_res;
}

static void ready_input(ErlDrvData drv_data, ErlDrvEvent event)
{
    our_data_t* data = (our_data_t*)drv_data;
    PGconn* conn = data->conn;
    ei_x_buff x;
    PGresult* res;

    PQconsumeInput(conn);
    if (PQisBusy(conn)) 
	return;
    ei_x_new_with_version(&x);
    res = PQgetResult(conn);
    encode_result(&x, res, conn);
    driver_output(data->port, x.buff, x.index);
    ei_x_free(&x);
    PQclear(res);
    for (;;) {
	res = PQgetResult(conn);
	if (res == NULL)
	    break;
	PQclear(res);
    }
}