*
* Copyright (c) 1994, Regents of the University of California
*
- * $Id: pqcomm.c,v 1.63 1999/01/17 06:18:26 momjian Exp $
+ * $Id: pqcomm.c,v 1.64 1999/01/23 22:27:28 tgl Exp $
*
*-------------------------------------------------------------------------
*/
/*
* INTERFACE ROUTINES
- * pq_init - initialize libpq
+ * pq_init - initialize libpq
* pq_getport - return the PGPORT setting
* pq_close - close input / output connections
* pq_flush - flush pending output
+ * pq_recvbuf - load some bytes into the input buffer
* pq_getstr - get a null terminated string from connection
- * pq_getchar - get 1 character from connection
- * pq_peekchar - peek at first character in connection
+ * pq_getchar - get 1 character from connection
+ * pq_peekchar - peek at next character from connection
* pq_getnchar - get n characters from connection, and null-terminate
* pq_getint - get an integer from connection
- * pq_putchar - send 1 character to connection
+ * pq_putchar - send 1 character to connection
* pq_putstr - send a null terminated string to connection
* pq_putnchar - send n characters to connection
* pq_putint - send an integer to connection
- * pq_putncharlen - send n characters to connection
+ * pq_putncharlen - send n characters to connection
* (also send an int header indicating
* the length)
* pq_getinaddr - initialize address from host and port number
* pq_getinserv - initialize address from host and service name
*
- * StreamDoUnlink - Shutdown UNIX socket connectioin
- * StreamServerPort - Open sock stream
- * StreamConnection - Create new connection with client
- * StreamClose - Close a client/backend connection
+ * StreamDoUnlink - Shutdown UNIX socket connection
+ * StreamServerPort - Open socket stream
+ * StreamConnection - Create new connection with client
+ * StreamClose - Close a client/backend connection
*
* NOTES
- * Frontend is now completey in interfaces/libpq, and no
- * functions from this file is used.
+ * Frontend is now completely in interfaces/libpq, and no
+ * functions from this file are used there.
*
*/
#include "postgres.h"
extern FILE * debug_port; /* in util.c */
+/*
+ * Buffers
+ */
+char PqSendBuffer[PQ_BUFFER_SIZE];
+char PqRecvBuffer[PQ_BUFFER_SIZE];
+int PqSendPointer,PqRecvPointer,PqRecvLength;
+
+
/* --------------------------------
* pq_init - open portal file descriptors
* --------------------------------
void
pq_init(int fd)
{
+ PqSendPointer = PqRecvPointer = PqRecvLength = 0;
PQnotifies_init();
if (getenv("LIBPQ_DEBUG"))
debug_port = stderr;
/* -------------------------
* pq_getchar()
*
- * get a character from the input file,
- *
+ * get a character from the input file, or EOF if trouble
+ * --------------------------------
*/
int
pq_getchar(void)
{
- char c;
-
- while (recv(MyProcPort->sock, &c, 1, 0) != 1) {
- if (errno != EINTR)
- return EOF; /* Not interrupted, so something went wrong */
+ while (PqRecvPointer >= PqRecvLength)
+ {
+ if (pq_recvbuf()) /* If nothing in buffer, then recv some */
+ return EOF; /* Failed to recv data */
}
-
- return c;
+ return PqRecvBuffer[PqRecvPointer++];
}
-/*
+/* -------------------------
+ * pq_peekchar()
+ *
+ * get a character from the connection, but leave it in the buffer
+ * to be read again
* --------------------------------
- * pq_peekchar - get 1 character from connection, but leave it in the stream
*/
-int
-pq_peekchar(void) {
- char c;
- while (recv(MyProcPort->sock, &c, 1, MSG_PEEK) != 1) {
- if (errno != EINTR)
- return EOF; /* Not interrupted, so something went wrong */
+int
+pq_peekchar(void)
+{
+ while (PqRecvPointer >= PqRecvLength)
+ {
+ if (pq_recvbuf()) /* If nothing in buffer, then recv some */
+ return EOF; /* Failed to recv data */
}
-
- return c;
+ /* Note we don't bump the pointer... */
+ return PqRecvBuffer[PqRecvPointer];
}
-
-
/* --------------------------------
* pq_getport - return the PGPORT setting
void
pq_close()
{
- close(MyProcPort->sock);
+ close(MyProcPort->sock);
PQnotifies_init();
}
/* --------------------------------
* pq_flush - flush pending output
+ *
+ * returns 0 if OK, EOF if trouble
* --------------------------------
*/
-void
+int
pq_flush()
{
- /* Not supported/required? */
+ char *bufptr = PqSendBuffer;
+ char *bufend = PqSendBuffer + PqSendPointer;
+
+ while (bufptr < bufend)
+ {
+ int r = send(MyProcPort->sock, bufptr, bufend - bufptr, 0);
+ if (r <= 0)
+ {
+ if (errno == EINTR)
+ continue; /* Ok if we were interrupted */
+ /* We would like to use elog() here, but cannot because elog
+ * tries to write to the client, which would cause a recursive
+ * flush attempt! So just write it out to the postmaster log.
+ */
+ fprintf(stderr, "pq_flush: send() failed, errno %d\n", errno);
+ /* We drop the buffered data anyway so that processing
+ * can continue, even though we'll probably quit soon.
+ */
+ PqSendPointer = 0;
+ return EOF;
+ }
+ bufptr += r;
+ }
+ PqSendPointer = 0;
+ return 0;
+}
+
+/* --------------------------------
+ * pq_recvbuf - load some bytes into the input buffer
+ *
+ * returns 0 if OK, EOF if trouble
+ * --------------------------------
+ */
+
+int
+pq_recvbuf()
+{
+ if (PqRecvPointer > 0)
+ {
+ if (PqRecvLength > PqRecvPointer)
+ {
+ /* still some unread data, left-justify it in the buffer */
+ memmove(PqRecvBuffer, PqRecvBuffer+PqRecvPointer,
+ PqRecvLength-PqRecvPointer);
+ PqRecvLength -= PqRecvPointer;
+ PqRecvPointer = 0;
+ }
+ else
+ PqRecvLength = PqRecvPointer = 0;
+ }
+
+ /* Can fill buffer from PqRecvLength and upwards */
+ for (;;)
+ {
+ int r = recv(MyProcPort->sock, PqRecvBuffer + PqRecvLength,
+ PQ_BUFFER_SIZE - PqRecvLength, 0);
+ if (r <= 0)
+ {
+ if (errno == EINTR)
+ continue; /* Ok if interrupted */
+ /* We would like to use elog() here, but dare not because elog
+ * tries to write to the client, which will cause problems
+ * if we have a hard communications failure ...
+ * So just write the message to the postmaster log.
+ */
+ fprintf(stderr, "pq_recvbuf: recv() failed, errno %d\n", errno);
+ return EOF;
+ }
+ /* r contains number of bytes read, so just incr length */
+ PqRecvLength += r;
+ return 0;
+ }
}
/* --------------------------------
int
pq_getnchar(char *s, int off, int maxlen)
{
- int r = pqGetNBytes(s + off, maxlen);
+ int r = pqGetNBytes(s + off, maxlen);
s[off+maxlen] = '\0';
return r;
}
if (setsockopt(port->sock, pe->p_proto, TCP_NODELAY,
&on, sizeof(on)) < 0)
{
- elog(ERROR, "postmaster: setsockopt failed");
+ elog(ERROR, "postmaster: setsockopt failed: %m");
return STATUS_ERROR;
}
}
*/
int pq_putchar(char c)
{
- char isDone = 0;
-
- do {
- if (send(MyProcPort->sock, &c, 1, 0) != 1) {
- if (errno != EINTR)
- return EOF; /* Anything other than interrupt is error! */
- }
- else
- isDone = 1; /* Done if we sent one char */
- } while (!isDone);
- return c;
+ if (PqSendPointer >= PQ_BUFFER_SIZE)
+ if (pq_flush()) /* If buffer is full, then flush it out */
+ return EOF;
+ PqSendBuffer[PqSendPointer++] = c; /* Put in buffer */
+ return c;
}
-
-
-
n = ((PG_PROTOCOL_MAJOR(FrontendProtocol) == 0) ? hton_l(integer) : htonl((uint32) integer));
#endif
- return pqPutNBytes((char *)&n,4);
+ return pqPutNBytes((char *)&n, 4);
}
/* --------------------------------------------------------------------- */
{
uint16 n;
- if (pqGetNBytes((char *)&n,2) != 0)
+ if (pqGetNBytes((char *)&n, 2) != 0)
return EOF;
#ifdef FRONTEND
}
/* --------------------------------------------------------------------- */
-/* pqGetNBytes: Read a chunk of exactly len bytes in buffer s (which must be 1
- byte longer) and terminate it with a '\0'.
- Return 0 if ok.
-*/
+/* pqGetNBytes: Read a chunk of exactly len bytes into buffer s.
+ * Return 0 if ok, EOF if trouble.
+ */
int
pqGetNBytes(char *s, size_t len)
{
- int bytesDone = 0;
-
- do {
- int r = recv(MyProcPort->sock, s+bytesDone, len-bytesDone, 0);
- if (r == 0 || r == -1) {
- if (errno != EINTR)
- return EOF; /* All other than signal-interrupted is error */
- continue; /* Otherwise, try again */
- }
-
- /* r contains number of bytes received */
- bytesDone += r;
-
- } while (bytesDone < len);
- /* Zero-termination now in pq_getnchar() instead */
+ size_t amount;
+
+ while (len > 0)
+ {
+ while (PqRecvPointer >= PqRecvLength)
+ {
+ if (pq_recvbuf()) /* If nothing in buffer, then recv some */
+ return EOF; /* Failed to recv data */
+ }
+ amount = PqRecvLength - PqRecvPointer;
+ if (amount > len)
+ amount = len;
+ memcpy(s, PqRecvBuffer + PqRecvPointer, amount);
+ PqRecvPointer += amount;
+ s += amount;
+ len -= amount;
+ }
return 0;
}
int
pqPutNBytes(const char *s, size_t len)
{
- int bytesDone = 0;
-
- do {
- int r = send(MyProcPort->sock, s+bytesDone, len-bytesDone, 0);
- if (r == 0 || r == -1) {
- if (errno != EINTR)
- return EOF; /* Only signal interruption allowed */
- continue; /* If interruped and read nothing, just try again */
- }
-
- /* r contains number of bytes sent so far */
- bytesDone += r;
- } while (bytesDone < len);
-
+ size_t amount;
+
+ while (len > 0)
+ {
+ if (PqSendPointer >= PQ_BUFFER_SIZE)
+ if (pq_flush()) /* If buffer is full, then flush it out */
+ return EOF;
+ amount = PQ_BUFFER_SIZE - PqSendPointer;
+ if (amount > len)
+ amount = len;
+ memcpy(PqSendBuffer + PqSendPointer, s, amount);
+ PqSendPointer += amount;
+ s += amount;
+ len -= amount;
+ }
return 0;
}
int c;
/*
- * Keep on reading until we get the terminating '\0' and discard those
- * bytes we don't have room for.
+ * Keep on reading until we get the terminating '\0',
+ * discarding any bytes we don't have room for.
*/
while ((c = pq_getchar()) != EOF && c != '\0')
{
return pqPutNBytes(s,strlen(s)+1);
}
-
*
* Copyright (c) 1994, Regents of the University of California
*
- * $Id: libpq.h,v 1.23 1999/01/12 12:49:52 scrappy Exp $
+ * $Id: libpq.h,v 1.24 1999/01/23 22:27:25 tgl Exp $
*
*-------------------------------------------------------------------------
*/
extern void pq_gettty(char *tp);
extern int pq_getport(void);
extern void pq_close(void);
-extern void pq_flush(void);
+extern int pq_flush(void);
+extern int pq_recvbuf(void);
extern int pq_getstr(char *s, int maxlen);
extern int PQgetline(char *s, int maxlen);
extern int PQputline(char *s);
-extern int pq_getchar(void);
-extern int pq_peekchar(void);
+extern int pq_getchar(void);
+extern int pq_peekchar(void);
extern int pq_getnchar(char *s, int off, int maxlen);
extern int pq_getint(int b);
extern int pq_putchar(char c);
extern int StreamConnection(int server_fd, Port *port);
extern void StreamClose(int sock);
+/*
+ * Internal send/receive buffers in libpq.
+ * These probably shouldn't be global at all, but unless we merge
+ * pqcomm.c and pqcomprim.c they have to be...
+ */
+
+#define PQ_BUFFER_SIZE 8192
+
+extern char PqSendBuffer[PQ_BUFFER_SIZE];
+extern int PqSendPointer; /* Next index to store a byte in PqSendBuffer */
+extern char PqRecvBuffer[PQ_BUFFER_SIZE];
+extern int PqRecvPointer; /* Next index to read a byte from PqRecvBuffer */
+extern int PqRecvLength; /* End of data available in PqRecvBuffer */
+
#endif /* LIBPQ_H */
projects / postgresql / commitdiff