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;
}

/*
 * Wait until we can read WAL stream, or timeout.
 *
 * Returns true if data has become available for reading, false if timed out
 * or interrupted by signal.
 *
 * This is based on pqSocketCheck.
 */
static bool
libpq_select(int timeout_ms)
{
	int			ret;

	Assert(streamConn != NULL);
	if (PQsocket(streamConn) < 0)
		ereport(ERROR,
				(errcode_for_socket_access(),
				 errmsg("socket not open")));

	/* We use poll(2) if available, otherwise select(2) */
	{
#ifdef HAVE_POLL
		struct pollfd input_fd;

		input_fd.fd = PQsocket(streamConn);
		input_fd.events = POLLIN | POLLERR;
		input_fd.revents = 0;

		ret = poll(&input_fd, 1, timeout_ms);
#else							/* !HAVE_POLL */

		fd_set		input_mask;
		struct timeval timeout;
		struct timeval *ptr_timeout;

		FD_ZERO(&input_mask);
		FD_SET(PQsocket(streamConn), &input_mask);

		if (timeout_ms < 0)
			ptr_timeout = NULL;
		else
		{
			timeout.tv_sec = timeout_ms / 1000;
			timeout.tv_usec = (timeout_ms % 1000) * 1000;
			ptr_timeout = &timeout;
		}

		ret = select(PQsocket(streamConn) + 1, &input_mask,
					 NULL, NULL, ptr_timeout);
#endif   /* HAVE_POLL */
	}

	if (ret == 0 || (ret < 0 && errno == EINTR))
		return false;
	if (ret < 0)
		ereport(ERROR,
				(errcode_for_socket_access(),
				 errmsg("select() failed: %m")));
	return true;
}

	bool DoConnect()
	{
		sql = PQconnectStart(GetDSN().c_str());
		if (!sql)
			return false;

		if(PQstatus(sql) == CONNECTION_BAD)
			return false;

		if(PQsetnonblocking(sql, 1) == -1)
			return false;

		/* OK, we've initalised the connection, now to get it hooked into the socket engine
		* and then start polling it.
		*/
		this->fd = PQsocket(sql);

		if(this->fd <= -1)
			return false;

		if (!ServerInstance->SE->AddFd(this, FD_WANT_NO_WRITE | FD_WANT_NO_READ))
		{
			ServerInstance->Logs->Log(MODNAME, LOG_DEBUG, "BUG: Couldn't add pgsql socket to socket engine");
			return false;
		}

		/* Socket all hooked into the engine, now to tell PgSQL to start connecting */
		return DoPoll();
	}

int
evpg_connect(struct evpg_cfg *config, const char *connstr)
{
    int pgsock;
    int status;
    struct evpg_db_node *dbnode;
    void **usrdata;

    if (!(dbnode = calloc(sizeof(struct evpg_db_node), 1)))
	return -1;

    dbnode->dbconn = PQconnectStart(connstr);
    
    pgsock = PQsocket(dbnode->dbconn);

    /* set this dbnode into an active state since it is not 
       ready to be used by the calling application */
    evpg_set_active(config, dbnode);

    /* we want to pass both our config, and our node */
    usrdata = malloc(sizeof(void *) * 2);
    usrdata[0] = config;
    usrdata[1] = dbnode;

    /* start the non-blocking connect event */
    event_set(&dbnode->event, pgsock, EV_WRITE, 
	    (void *)evpg_connect_check, usrdata);
    event_add(&dbnode->event, 0);
}

SWITCH_DECLARE(switch_pgsql_status_t) switch_pgsql_handle_connect(switch_pgsql_handle_t *handle)
{
#ifdef SWITCH_HAVE_PGSQL
    if (handle->state == SWITCH_PGSQL_STATE_CONNECTED) {
        switch_pgsql_handle_disconnect(handle);
        switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG1, "Re-connecting %s\n", handle->dsn);
    }

    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG1, "Connecting %s\n", handle->dsn);

    handle->con = PQconnectdb(handle->dsn);
    if (PQstatus(handle->con) != CONNECTION_OK) {
        char *err_str;
        if ((err_str = switch_pgsql_handle_get_error(handle))) {
            switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "%s\n", err_str);
            switch_safe_free(err_str);
        } else {
            switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Failed to connect to the database [%s]\n", handle->dsn);
            switch_pgsql_handle_disconnect(handle);
        }
        return SWITCH_PGSQL_FAIL;
    }

    switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG1, "Connected to [%s]\n", handle->dsn);
    handle->state = SWITCH_PGSQL_STATE_CONNECTED;
    handle->sock = PQsocket(handle->con);
    return SWITCH_PGSQL_SUCCESS;
#else
    return SWITCH_PGSQL_FAIL;
#endif
}

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);
}

/*
 * set_connection_status_bad does not remove the given connection from the connection hash.
 * It simply shuts down the underlying socket. On success, it returns true.
 */
Datum
set_connection_status_bad(PG_FUNCTION_ARGS)
{
	char *nodeName = PG_GETARG_CSTRING(0);
	int32 nodePort = PG_GETARG_INT32(1);
	int socket = -1;
	int shutdownStatus = 0;
	int pqStatus PG_USED_FOR_ASSERTS_ONLY = 0;

	PGconn *connection = GetOrEstablishConnection(nodeName, nodePort);
	if (connection == NULL)
	{
		PG_RETURN_BOOL(false);
	}

	/* Prevent further reads/writes... */
	socket = PQsocket(connection);
	shutdownStatus = shutdown(socket, SHUT_RDWR);
	if (shutdownStatus != 0)
	{
		ereport(ERROR, (errcode_for_socket_access(), errmsg("shutdown failed")));
	}

	/* ... and make libpq notice by reading data. */
	pqStatus = PQconsumeInput(connection);

	Assert(pqStatus == 0); /* expect failure */

	PG_RETURN_BOOL(true);
}

/*
 * MultiClientRegisterWait adds a connection to be waited upon, waiting for
 * executionStatus.
 */
void
MultiClientRegisterWait(WaitInfo *waitInfo, TaskExecutionStatus executionStatus,
						int32 connectionId)
{
	PGconn *connection = NULL;
	struct pollfd *pollfd = NULL;

	Assert(waitInfo->registeredWaiters < waitInfo->maxWaiters);

	if (executionStatus == TASK_STATUS_READY)
	{
		waitInfo->haveReadyWaiter = true;
		return;
	}
	else if (executionStatus == TASK_STATUS_ERROR)
	{
		waitInfo->haveFailedWaiter = true;
		return;
	}

	connection = ClientConnectionArray[connectionId];
	pollfd = &waitInfo->pollfds[waitInfo->registeredWaiters];
	pollfd->fd = PQsocket(connection);
	if (executionStatus == TASK_STATUS_SOCKET_READ)
	{
		pollfd->events = POLLERR | POLLIN;
	}
	else if (executionStatus == TASK_STATUS_SOCKET_WRITE)
	{
		pollfd->events = POLLERR | POLLOUT;
	}
	waitInfo->registeredWaiters++;
}

void
PgStartNotifyEventSource(Pg_ConnectionId * connid)
{
	/* Start the notify event source if it isn't already running */
	if (!connid->notifier_running)
	{
		int			pqsock = PQsocket(connid->conn);

		if (pqsock >= 0)
		{
#if TCL_MAJOR_VERSION >= 8
			Tcl_CreateChannelHandler(connid->notifier_channel,
									 TCL_READABLE,
									 Pg_Notify_FileHandler,
									 (ClientData) connid);
#else
			/* In Tcl 7.5 and 7.6, we need to gin up a Tcl_File. */
			Tcl_File	tclfile = Tcl_GetFile((ClientData) pqsock, TCL_UNIX_FD);

			Tcl_CreateFileHandler(tclfile, TCL_READABLE,
							 Pg_Notify_FileHandler, (ClientData) connid);
			connid->notifier_socket = pqsock;
#endif
			connid->notifier_running = 1;
		}
	}
}

void
PgNotifyTransferEvents(Pg_ConnectionId * connid)
{
	PGnotify   *notify;

	while ((notify = PQnotifies(connid->conn)) != NULL)
	{
		NotifyEvent *event = (NotifyEvent *) ckalloc(sizeof(NotifyEvent));

		event->header.proc = Pg_Notify_EventProc;
		event->notify = notify;
		event->connid = connid;
		Tcl_QueueEvent((Tcl_Event *) event, TCL_QUEUE_TAIL);
	}

	/*
	 * This is also a good place to check for unexpected closure of the
	 * connection (ie, backend crash), in which case we must shut down the
	 * notify event source to keep Tcl from trying to select() on the now-
	 * closed socket descriptor.  But don't kill on-connection-loss
	 * events; in fact, register one.
	 */
	if (PQsocket(connid->conn) < 0)
		PgConnLossTransferEvents(connid);
}

static void
init_slot(ParallelSlot *slot, PGconn *conn)
{
	slot->connection = conn;
	slot->isFree = true;
	slot->sock = PQsocket(conn);
}

struct connection_struct* initDatabase(struct event_base* base) {
  struct connection_struct* database = malloc(sizeof(struct connection_struct));
  database->query_count = 0;
  database->report_errors = 0;
  database->queries = NULL;
  database->last_query = NULL;
  database->conn = PQconnectdb(db_connect);
  if (PQstatus(database->conn) != CONNECTION_OK) {
    fprintf(stderr, "%s\n", PQerrorMessage(database->conn));
    PQfinish(database->conn);
    exit(1);
  } else
    PQsetnonblocking(database->conn, 1);
  struct event* event = event_new(base, PQsocket(database->conn), EV_READ|EV_PERSIST, pq_event, database);
  event_add(event, NULL);
  if (all_databases == NULL) {
    all_databases = malloc(sizeof(struct database_list));
    memset(all_databases, 0, sizeof(struct database_list));
    all_databases->db = database;
  } else {
    struct database_list* node = all_databases;
    while (node->next)
      node = node->next;
    node->next = malloc(sizeof(struct database_list));
    memset(node->next, 0, sizeof(struct database_list));
    node->next->db = database;
  }
  return database;
};

static void
wait_for_flush(PlxFn *plx_fn, PGconn *pq_conn)
{
    struct epoll_event listenev;
    struct epoll_event event;
    int                res;

    res = PQflush(pq_conn);
    if (!res)
        return;
    if (res == -1)
        plx_error(plx_fn, "PQflush error %s", PQerrorMessage(pq_conn));

    listenev.events = EPOLLOUT;
    listenev.data.fd = PQsocket(pq_conn);

    if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listenev.data.fd, &listenev) < 0)
        plx_error(plx_fn, "epoll: socket adding failed");

    while (res)
    {
        CHECK_FOR_INTERRUPTS();
        epoll_wait(epoll_fd, &event, 1, 1);
        res = PQflush(pq_conn);
        if (res == -1)
        {
            epoll_ctl(epoll_fd, EPOLL_CTL_DEL, listenev.data.fd, &listenev);
            plx_error(plx_fn, "%s", PQerrorMessage(pq_conn));
        }
    }
    epoll_ctl(epoll_fd, EPOLL_CTL_DEL, listenev.data.fd, &listenev);
}

/*
 * Create and register a new channel for the connection
 */
void
PgSetConnectionId(Tcl_Interp *interp, PGconn *conn)
{
	Tcl_Channel conn_chan;
	Pg_ConnectionId *connid;
	int			i;

	connid = (Pg_ConnectionId *) ckalloc(sizeof(Pg_ConnectionId));
	connid->conn = conn;
	connid->res_count = 0;
	connid->res_last = -1;
	connid->res_max = RES_START;
	connid->res_hardmax = RES_HARD_MAX;
	connid->res_copy = -1;
	connid->res_copyStatus = RES_COPY_NONE;
	connid->results = (PGresult **) ckalloc(sizeof(PGresult *) * RES_START);
	for (i = 0; i < RES_START; i++)
		connid->results[i] = NULL;
	connid->notify_list = NULL;
	connid->notifier_running = 0;

	sprintf(connid->id, "pgsql%d", PQsocket(conn));

#if TCL_MAJOR_VERSION >= 8
	connid->notifier_channel = Tcl_MakeTcpClientChannel((ClientData) PQsocket(conn));
	Tcl_RegisterChannel(NULL, connid->notifier_channel);
#else
	connid->notifier_socket = -1;
#endif

#if TCL_MAJOR_VERSION == 7 && TCL_MINOR_VERSION == 5
	/* Original signature (only seen in Tcl 7.5) */
	conn_chan = Tcl_CreateChannel(&Pg_ConnType, connid->id, NULL, NULL, (ClientData) connid);
#else
	/* Tcl 7.6 and later use this */
	conn_chan = Tcl_CreateChannel(&Pg_ConnType, connid->id, (ClientData) connid,
								  TCL_READABLE | TCL_WRITABLE);
#endif

	Tcl_SetChannelOption(interp, conn_chan, "-buffering", "line");
	Tcl_SetResult(interp, connid->id, TCL_VOLATILE);
	Tcl_RegisterChannel(interp, conn_chan);
}

static PyObject *
psyco_conn_fileno(connectionObject *self)
{
    long int socket;

    EXC_IF_CONN_CLOSED(self);

    socket = (long int)PQsocket(self->pgconn);

    return PyInt_FromLong(socket);
}

/* Blocks until more data is received from the server. You don't have to use
 * this if you have your own select loop. */
int db_client_wait(client_context_t context) {
    fd_set input_mask;
    FD_ZERO(&input_mask);

    int rep_fd = PQsocket(context->repl.conn);
    int max_fd = rep_fd;
    FD_SET(rep_fd, &input_mask);

    if (context->sql_conn) {
        int sql_fd = PQsocket(context->sql_conn);
        if (sql_fd > max_fd) max_fd = sql_fd;
        FD_SET(sql_fd, &input_mask);
    }

    struct timeval timeout;
    timeout.tv_sec = 1;
    timeout.tv_usec = 0;

    int ret = select(max_fd + 1, &input_mask, NULL, NULL, &timeout);

    if (ret == 0 || (ret < 0 && errno == EINTR)) {
        return 0; /* timeout or signal */
    }
    if (ret < 0) {
        client_error(context, "select() failed: %s", strerror(errno));
        return errno;
    }

    /* Data has arrived on the socket */
    if (!PQconsumeInput(context->repl.conn)) {
        client_error(context, "Could not receive replication data: %s",
                PQerrorMessage(context->repl.conn));
        return EIO;
    }
    if (context->sql_conn && !PQconsumeInput(context->sql_conn)) {
        client_error(context, "Could not receive snapshot data: %s",
                PQerrorMessage(context->sql_conn));
        return EIO;
    }
    return 0;
}

static void
init_slot(ParallelSlot *slot, PGconn *conn, const char *progname)
{
	slot->connection = conn;
	slot->isFree = true;
	slot->sock = PQsocket(conn);

	if (slot->sock < 0)
	{
		fprintf(stderr, _("%s: invalid socket: %s"), progname,
				PQerrorMessage(conn));
		exit(1);
	}
}

static void 
evpg_make_evquery(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];

    if (!dbnode)
    {
       	if (!(dbnode = evpg_snatch_connection(config)))
	{
	    event_set(event, 0, EV_WRITE, (void *)evpg_make_evquery, data);
	    event_add(event, 0);
	    return;
	}
    }

    PQsendQuery(dbnode->dbconn, querystr);

    if (PQstatus(dbnode->dbconn) != CONNECTION_OK)
    {
	cb(NULL, usrdata);
	evpg_set_ready(config, dbnode);
	free(event);
	free(data);
    }

    data[4] = dbnode;

    evpg_set_active(config, dbnode);

    event_set(&dbnode->event, PQsocket(dbnode->dbconn), EV_WRITE, 
	    (void *)evpg_query_finished, data);
    event_add(&dbnode->event, 0);
}

static PGresult*
wait_for_result(PlxFn *plx_fn, PlxConn *plx_conn)
{
    struct epoll_event  listenev;
    struct epoll_event  event;
    PGconn             *pq_conn   = plx_conn->pq_conn;
    PGresult           *pg_result = NULL;

    listenev.events = EPOLLIN;
    listenev.data.fd = PQsocket(pq_conn);

    if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listenev.data.fd, &listenev) < 0)
        plx_error(plx_fn, "epoll: socket adding failed");

    PG_TRY();
    {
        int tmp;

        while ((tmp = is_pq_busy(pq_conn)))
        {
            if (tmp == -1)
            {
                epoll_ctl(epoll_fd, EPOLL_CTL_DEL, listenev.data.fd, &listenev);
                plx_error(plx_fn, "%s", PQerrorMessage(pq_conn));
            }
            CHECK_FOR_INTERRUPTS();
            epoll_wait(epoll_fd, &event, 1, 10000);
        }
    }
    PG_CATCH();
    {
        epoll_ctl(epoll_fd, EPOLL_CTL_DEL, listenev.data.fd, &listenev);

        if (geterrcode() == ERRCODE_QUERY_CANCELED)
            PQrequestCancel(pq_conn);
        pg_result = PQgetResult(pq_conn);
        if (pg_result)
            PQclear(pg_result);
        PG_RE_THROW();
    }
    PG_END_TRY();

    epoll_ctl(epoll_fd, EPOLL_CTL_DEL, listenev.data.fd, &listenev);
    return PQgetResult(pq_conn);
}

    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);
        }
      }
    }