Bugzilla bug #17223: implemented the PR_MSG_PEEK flag for PR_Recv() for

BeOS, Mac OS, OS/2, and Windows.  On BeOS, Mac OS, and NT, PR_MSG_PEEK
is emulated with a peek buffer in the NSPR file descriptors.
Modified files: _beos.h, _macos.h, _winnt.h, primpl.h, prsocket.c,
macsockotpt.c, os2sock.c, ntio.c, and w95sock.c

9 files changed, 117 insertions, 13 deletions

diff --git a/pr/include/md/_beos.h b/pr/include/md/_beos.h
index 8b56dec3..cd8eee65 100644
--- a/pr/include/md/_beos.h
+++ b/pr/include/md/_beos.h
@@ -68,6 +68,9 @@
 #define _PR_GLOBAL_THREADS_ONLY
 #define _PR_BTHREADS
 #define _PR_NEED_FAKE_POLL
+#define _PR_HAVE_PEEK_BUFFER
+#define _PR_PEEK_BUFFER_MAX (16 * 1024)
+#define _PR_FD_NEED_EMULATE_MSG_PEEK(fd) 1
 
 /* Define threading functions and objects as native BeOS */
 struct _MDThread {
diff --git a/pr/include/md/_macos.h b/pr/include/md/_macos.h
index d01fd2c1..8225c510 100644
--- a/pr/include/md/_macos.h
+++ b/pr/include/md/_macos.h
@@ -46,7 +46,9 @@
 #include <setjmp.h>
 #include <Errors.h>
 
-#define _PR_EMULATE_MSG_PEEK
+#define _PR_HAVE_PEEK_BUFFER
+#define _PR_PEEK_BUFFER_MAX (16 * 1024)
+#define _PR_FD_NEED_EMULATE_MSG_PEEK(fd) 1
 
 struct _MDProcess {
     PRInt8 notused;
diff --git a/pr/include/md/_winnt.h b/pr/include/md/_winnt.h
index 4ea950ba..b4bd27f2 100644
--- a/pr/include/md/_winnt.h
+++ b/pr/include/md/_winnt.h
@@ -50,7 +50,10 @@
 #define _PR_HAVE_ATOMIC_OPS
 #define _PR_HAVE_ATOMIC_CAS
 #define PR_HAVE_WIN32_NAMED_SHARED_MEMORY
-#define _PR_EMULATE_MSG_PEEK
+#define _PR_HAVE_PEEK_BUFFER
+#define _PR_PEEK_BUFFER_MAX (32 * 1024)
+#define _PR_FD_NEED_EMULATE_MSG_PEEK(fd) \
+    (!(fd)->secret->nonblocking && !(fd)->secret->inheritable)
 
 /* --- Common User-Thread/Native-Thread Definitions --------------------- */
 
diff --git a/pr/include/private/primpl.h b/pr/include/private/primpl.h
index c9d62812..f87adcce 100644
--- a/pr/include/private/primpl.h
+++ b/pr/include/private/primpl.h
@@ -1577,8 +1577,8 @@ struct PRFilePrivate {
     PRBool inheritable;
     PRFileDesc *next;
     PRIntn lockCount;
-#ifdef _PR_EMULATE_MSG_PEEK
-    char *peekBuf;
+#ifdef _PR_HAVE_PEEK_BUFFER
+    char *peekBuffer;
     PRInt32 peekBufSize;
     PRInt32 peekBytes;
 #endif
diff --git a/pr/src/io/prsocket.c b/pr/src/io/prsocket.c
index 186ea0a6..57a1cee3 100644
--- a/pr/src/io/prsocket.c
+++ b/pr/src/io/prsocket.c
@@ -543,6 +543,10 @@ PRIntervalTime timeout)
 	PRInt32 rv;
 	PRThread *me = _PR_MD_CURRENT_THREAD();
 
+	if ((flags != 0) && (flags != PR_MSG_PEEK)) {
+		PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);
+		return -1;
+	}
 	if (_PR_PENDING_INTERRUPT(me)) {
 		me->flags &= ~_PR_INTERRUPT;
 		PR_SetError(PR_PENDING_INTERRUPT_ERROR, 0);
@@ -553,11 +557,61 @@ PRIntervalTime timeout)
 		return -1;
 	}
 
-	PR_LOG(_pr_io_lm, PR_LOG_MAX, ("recv: fd=%p osfd=%d buf=%p amount=%d",
-		    						fd, fd->secret->md.osfd, buf, amount));
+	PR_LOG(_pr_io_lm, PR_LOG_MAX, ("recv: fd=%p osfd=%d buf=%p amount=%d flags=%d",
+		    						fd, fd->secret->md.osfd, buf, amount, flags));
+
+#ifdef _PR_HAVE_PEEK_BUFFER
+	if (fd->secret->peekBytes != 0) {
+		rv = (amount < fd->secret->peekBytes) ?
+			amount : fd->secret->peekBytes;
+		memcpy(buf, fd->secret->peekBuffer, rv);
+		if (flags == 0) {
+			/* consume the bytes in the peek buffer */
+			fd->secret->peekBytes -= rv;
+			if (fd->secret->peekBytes != 0) {
+				memmove(fd->secret->peekBuffer,
+					fd->secret->peekBuffer + rv,
+					fd->secret->peekBytes);
+			}
+		}
+		return rv;
+	}
+
+	/* allocate peek buffer, if necessary */
+	if ((PR_MSG_PEEK == flags) && _PR_FD_NEED_EMULATE_MSG_PEEK(fd)) {
+		PR_ASSERT(0 == fd->secret->peekBytes);
+		/* impose a max size on the peek buffer */
+		if (amount > _PR_PEEK_BUFFER_MAX) {
+			amount = _PR_PEEK_BUFFER_MAX;
+		}
+		if (fd->secret->peekBufSize < amount) {
+			if (fd->secret->peekBuffer) {
+				PR_Free(fd->secret->peekBuffer);
+			}
+			fd->secret->peekBufSize = amount;
+			fd->secret->peekBuffer = PR_Malloc(amount);
+			if (NULL == fd->secret->peekBuffer) {
+				fd->secret->peekBufSize = 0;
+				PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);
+				return -1;
+			}
+		}
+	}
+#endif
+
 	rv = _PR_MD_RECV(fd, buf, amount, flags, timeout);
 	PR_LOG(_pr_io_lm, PR_LOG_MAX, ("recv -> %d, error = %d, os error = %d",
 		rv, PR_GetError(), PR_GetOSError()));
+
+#ifdef _PR_HAVE_PEEK_BUFFER
+	if ((PR_MSG_PEEK == flags) && _PR_FD_NEED_EMULATE_MSG_PEEK(fd)) {
+		if (rv > 0) {
+			memcpy(fd->secret->peekBuffer, buf, me->md.blocked_io_bytes);
+			fd->secret->peekBytes = me->md.blocked_io_bytes;
+		}
+	}
+#endif
+
 	return rv;
 }
 
@@ -626,6 +680,15 @@ static PRStatus PR_CALLBACK SocketClose(PRFileDesc *fd)
 		fd->secret->state = _PR_FILEDESC_CLOSED;
 	}
 
+#ifdef _PR_HAVE_PEEK_BUFFER
+	if (fd->secret->peekBuffer) {
+		PR_ASSERT(fd->secret->peekBufSize > 0);
+		PR_DELETE(fd->secret->peekBuffer);
+		fd->secret->peekBufSize = 0;
+		fd->secret->peekBytes = 0;
+	}
+#endif
+
 	PR_FreeFileDesc(fd);
 	return PR_SUCCESS;
 }
@@ -633,6 +696,11 @@ static PRStatus PR_CALLBACK SocketClose(PRFileDesc *fd)
 static PRInt32 PR_CALLBACK SocketAvailable(PRFileDesc *fd)
 {
 	PRInt32 rv;
+#ifdef _PR_HAVE_PEEK_BUFFER
+	if (fd->secret->peekBytes != 0) {
+		return fd->secret->peekBytes;
+	}
+#endif
 	rv =  _PR_MD_SOCKETAVAILABLE(fd);
 	return rv;		
 }
@@ -640,6 +708,12 @@ static PRInt32 PR_CALLBACK SocketAvailable(PRFileDesc *fd)
 static PRInt64 PR_CALLBACK SocketAvailable64(PRFileDesc *fd)
 {
     PRInt64 rv;
+#ifdef _PR_HAVE_PEEK_BUFFER
+    if (fd->secret->peekBytes != 0) {
+        LL_I2L(rv, fd->secret->peekBytes);
+        return rv;
+    }
+#endif
     LL_I2L(rv, _PR_MD_SOCKETAVAILABLE(fd));
 	return rv;		
 }
diff --git a/pr/src/md/mac/macsockotpt.c b/pr/src/md/mac/macsockotpt.c
index 7580bc5d..f6669344 100644
--- a/pr/src/md/mac/macsockotpt.c
+++ b/pr/src/md/mac/macsockotpt.c
@@ -1349,7 +1349,8 @@ static PRInt32 SendReceiveStream(PRFileDesc *fd, void *buf, PRInt32 amount,
     PRThread *me = _PR_MD_CURRENT_THREAD();
     PRInt32 bytesLeft = amount;
 
-    PR_ASSERT(flags == 0);
+    PR_ASSERT(flags == 0 ||
+        (opCode == kSTREAM_RECEIVE && flags == PR_MSG_PEEK));
     PR_ASSERT(opCode == kSTREAM_SEND || opCode == kSTREAM_RECEIVE);
     
     if (endpoint == NULL) {
diff --git a/pr/src/md/os2/os2sock.c b/pr/src/md/os2/os2sock.c
index 34f5ff43..0d086fad 100644
--- a/pr/src/md/os2/os2sock.c
+++ b/pr/src/md/os2/os2sock.c
@@ -320,8 +320,15 @@ _PR_MD_RECV(PRFileDesc *fd, void *buf, PRInt32 amount, PRIntn flags,
     int socks[1]; 
     long lTimeout = -1; 
 #endif
+    int osflags;
 
-    while ((rv = recv( osfd, buf, amount, 0)) == -1) 
+    if (0 == flags) {
+        osflags = 0;
+    } else {
+        PR_ASSERT(PR_MSG_PEEK == flags);
+        osflags = MSG_PEEK;
+    }
+    while ((rv = recv( osfd, buf, amount, osflags)) == -1) 
     {
         if (((err = sock_errno()) == EWOULDBLOCK) 
             && (!fd->secret->nonblocking))
diff --git a/pr/src/md/windows/ntio.c b/pr/src/md/windows/ntio.c
index 20f0e695..d353a4a5 100644
--- a/pr/src/md/windows/ntio.c
+++ b/pr/src/md/windows/ntio.c
@@ -96,7 +96,7 @@ PRInt32 IsFileLocal(HANDLE hFile);
 static PRInt32 _md_MakeNonblock(HANDLE);
 
 PRInt32 _nt_nonblock_accept(PRFileDesc *fd, struct sockaddr_in *addr, int *len, PRIntervalTime);
-PRInt32 _nt_nonblock_recv(PRFileDesc *fd, char *buf, int len, PRIntervalTime);
+PRInt32 _nt_nonblock_recv(PRFileDesc *fd, char *buf, int len, int flags, PRIntervalTime);
 PRInt32 _nt_nonblock_send(PRFileDesc *fd, char *buf, int len, PRIntervalTime);
 PRInt32 _nt_nonblock_writev(PRFileDesc *fd, const PRIOVec *iov, int size, PRIntervalTime);
 PRInt32 _nt_nonblock_sendto(PRFileDesc *, const char *, int, const struct sockaddr *, int, PRIntervalTime);
@@ -1729,7 +1729,7 @@ _PR_MD_RECV(PRFileDesc *fd, void *buf, PRInt32 amount, PRIntn flags,
             PR_ASSERT(0 != rv);
             fd->secret->md.io_model_committed = PR_TRUE;
         }
-        return _nt_nonblock_recv(fd, buf, amount, timeout);
+        return _nt_nonblock_recv(fd, buf, amount, flags, timeout);
     }
 
     if (me->io_suspended) {
@@ -3808,14 +3808,21 @@ retry:
     return(rv);
 }
 
-PRInt32 _nt_nonblock_recv(PRFileDesc *fd, char *buf, int len, PRIntervalTime timeout)
+PRInt32 _nt_nonblock_recv(PRFileDesc *fd, char *buf, int len, int flags, PRIntervalTime timeout)
 {
     PRInt32 osfd = fd->secret->md.osfd;
     PRInt32 rv, err;
     struct timeval tv, *tvp;
     fd_set rd;
+    int osflags;
 
-    while ((rv = recv(osfd,buf,len,0)) == -1) {
+    if (0 == flags) {
+        osflags = 0;
+    } else {
+        PR_ASSERT(PR_MSG_PEEK == flags);
+        osflags = MSG_PEEK;
+    }
+    while ((rv = recv(osfd,buf,len,osflags)) == -1) {
         if (((err = WSAGetLastError()) == WSAEWOULDBLOCK)
                 && (!fd->secret->nonblocking)) {
             FD_ZERO(&rd);
diff --git a/pr/src/md/windows/w95sock.c b/pr/src/md/windows/w95sock.c
index 46b54843..a61005cf 100644
--- a/pr/src/md/windows/w95sock.c
+++ b/pr/src/md/windows/w95sock.c
@@ -171,8 +171,15 @@ _PR_MD_RECV(PRFileDesc *fd, void *buf, PRInt32 amount, PRIntn flags,
 {
     PRInt32 osfd = fd->secret->md.osfd;
     PRInt32 rv, err;
+    int osflags;
 
-    while ((rv = recv( osfd, buf, amount, 0)) == -1) 
+    if (0 == flags) {
+        osflags = 0;
+    } else {
+        PR_ASSERT(PR_MSG_PEEK == flags);
+        osflags = MSG_PEEK;
+    }
+    while ((rv = recv( osfd, buf, amount, osflags)) == -1) 
     {
         if (((err = WSAGetLastError()) == WSAEWOULDBLOCK) 
             && (!fd->secret->nonblocking))