7074436: (sc) SocketChannel can do short gathering writes when channel configured blocking (win)

Reviewed-by: chegar

src/windows/native/sun/nio/ch/SocketDispatcher.c

@@ -192,45 +192,66 @@ Java_sun_nio_ch_SocketDispatcher_writev0(JNIEnv *env, jclass clazz,
                                          jobject fdo, jlong address, jint len)
 {
     /* set up */
-    int i = 0;
+    int next_index, next_offset, ret=0;
     DWORD written = 0;
     jint fd = fdval(env, fdo);
     struct iovec *iovp = (struct iovec *)address;
     WSABUF *bufs = malloc(len * sizeof(WSABUF));
-    jint rem = MAX_BUFFER_SIZE;
+    jlong count = 0;
 
     if (bufs == 0) {
         JNU_ThrowOutOfMemoryError(env, 0);
         return IOS_THROWN;
     }
 
-    /* copy iovec into WSABUF */
-    for(i=0; i<len; i++) {
-        jint iov_len = iovp[i].iov_len;
-        if (iov_len > rem)
-            iov_len = rem;
-        bufs[i].buf = (char *)iovp[i].iov_base;
-        bufs[i].len = (u_long)iov_len;
-        rem -= iov_len;
-        if (rem == 0) {
-            len = i+1;
+    // next buffer and offset to consume
+    next_index = 0;
+    next_offset = 0;
+
+    while (next_index  < len) {
+        DWORD buf_count = 0;
+
+        /* Prepare the WSABUF array to a maximum total size of MAX_BUFFER_SIZE */
+        jint rem = MAX_BUFFER_SIZE;
+        while (next_index < len && rem > 0) {
+            jint iov_len = iovp[next_index].iov_len - next_offset;
+            char* ptr = (char *)iovp[next_index].iov_base;
+            ptr += next_offset;
+            if (iov_len > rem) {
+                iov_len = rem;
+                next_offset += rem;
+            } else {
+                next_index ++;
+                next_offset = 0;
+            }
+
+            bufs[buf_count].buf = ptr;
+            bufs[buf_count].len = (u_long)iov_len;
+            buf_count++;
+
+            rem -= iov_len;
+        }
+
+        /* write the buffers */
+        ret = WSASend((SOCKET)fd,           /* Socket */
+                              bufs,         /* pointers to the buffers */
+                              buf_count,    /* number of buffers to process */
+                              &written,     /* receives number of bytes written */
+                              0,            /* no flags */
+                              0,            /* no overlapped sockets */
+                              0);           /* no completion routine */
+
+        if (ret == SOCKET_ERROR) {
             break;
         }
-    }
 
-    /* read into the buffers */
-    i = WSASend((SOCKET)fd, /* Socket */
-            bufs,           /* pointers to the buffers */
-            (DWORD)len,     /* number of buffers to process */
-            &written,       /* receives number of bytes written */
-            0,              /* no flags */
-            0,              /* no overlapped sockets */
-            0);             /* no completion routine */
+        count += written;
+    }
 
     /* clean up */
     free(bufs);
 
-    if (i != 0) {
+    if (ret == SOCKET_ERROR && count == 0) {
         int theErr = (jint)WSAGetLastError();
         if (theErr == WSAEWOULDBLOCK) {
             return IOS_UNAVAILABLE;
@@ -239,7 +260,7 @@ Java_sun_nio_ch_SocketDispatcher_writev0(JNIEnv *env, jclass clazz,
         return IOS_THROWN;
     }
 
-    return convertLongReturnVal(env, (jlong)written, JNI_FALSE);
+    return convertLongReturnVal(env, count, JNI_FALSE);
 }
 
 JNIEXPORT void JNICALL

test/java/nio/channels/SocketChannel/ShortWrite.java

@@ -22,7 +22,7 @@
  */
 
 /* @test
- * @bug 7176630
+ * @bug 7176630 7074436
  * @summary Check for short writes on SocketChannels configured in blocking mode
  */
 
@@ -38,11 +38,12 @@ public class ShortWrite {
     static final Random rand = new Random();
 
     /**
-     * Returns a checksum on the remaining bytes in the given buffer.
+     * Returns a checksum on the remaining bytes in the given buffers.
      */
-    static long computeChecksum(ByteBuffer bb) {
+    static long computeChecksum(ByteBuffer... bufs) {
         CRC32 crc32 = new CRC32();
-        crc32.update(bb);
+        for (int i=0; i<bufs.length; i++)
+            crc32.update(bufs[i]);
         return crc32.getValue();
     }
 
@@ -71,15 +72,15 @@ public class ShortWrite {
     }
 
     /**
-     * Run test with a write of the given number of bytes.
+     * Exercise write(ByteBuffer) with given number of bytes.
      */
-    static void test(ExecutorService pool,
-                     SocketChannel source,
-                     SocketChannel sink,
-                     int size)
+    static void test1(ExecutorService pool,
+                      SocketChannel source,
+                      SocketChannel sink,
+                      int size)
         throws Exception
     {
-        System.out.println(size);
+        System.out.println("write(ByteBuffer), size=" + size);
 
         // random bytes in the buffer
         ByteBuffer buf = ByteBuffer.allocate(size);
@@ -101,6 +102,47 @@ public class ShortWrite {
             throw new RuntimeException("Checksum did not match");
     }
 
+    /**
+     * Exercise write(ByteBuffer[]) with buffers of the given sizes.
+     */
+    static void testN(ExecutorService pool,
+                      SocketChannel source,
+                      SocketChannel sink,
+                      int... sizes)
+        throws Exception
+    {
+        System.out.print("write(ByteBuffer[]), sizes=");
+        for (int size: sizes)
+            System.out.print(size + " ");
+        System.out.println();
+
+        int total = 0;
+        int len = sizes.length;
+        ByteBuffer[] bufs = new ByteBuffer[len];
+        for (int i=0; i<len; i++) {
+            int size = sizes[i];
+            ByteBuffer buf = ByteBuffer.allocate(size);
+            rand.nextBytes(buf.array());
+            bufs[i] = buf;
+            total += size;
+        }
+
+        // submit task to read the bytes
+        Future<Long> result = pool.submit(new Reader(sink, total));
+
+        // write the bytes
+        long n = source.write(bufs);
+        if (n != total)
+            throw new RuntimeException("Short write detected");
+
+        // check the bytes that were received match
+        for (int i=0; i<len; i++)
+            bufs[i].rewind();
+        long expected = computeChecksum(bufs);
+        long actual = result.get();
+        if (actual != expected)
+            throw new RuntimeException("Checksum did not match");
+    }
 
     public static void main(String[] args) throws Exception {
         ExecutorService pool = Executors.newSingleThreadExecutor();
@@ -114,17 +156,47 @@ public class ShortWrite {
                 try (SocketChannel source = SocketChannel.open(sa);
                      SocketChannel sink = ssc.accept())
                 {
-                    // run tests on sizes around 128k as that is the problem
-                    // area on Windows.
+                    // Exercise write(BufferBuffer) on sizes around 128k
                     int BOUNDARY = 128 * 1024;
                     for (int size=(BOUNDARY-2); size<=(BOUNDARY+2); size++) {
-                        test(pool, source, sink, size);
+                        test1(pool, source, sink, size);
                     }
 
-                    // run tests on random sizes
+                    // Exercise write(BufferBuffer) on random sizes
                     for (int i=0; i<20; i++) {
                         int size = rand.nextInt(1024*1024);
-                        test(pool, source, sink, size);
+                        test1(pool, source, sink, size);
+                    }
+
+                    // Exercise write(BufferBuffer[]) on sizes around 128k
+                    for (int i=BOUNDARY-2; i<=BOUNDARY+2; i++) {
+                        testN(pool, source, sink, i);
+                        testN(pool, source, sink, 0, i);
+                        testN(pool, source, sink, i, 0);
+                        for (int j=BOUNDARY-2; j<=BOUNDARY+2; j++) {
+                            testN(pool, source, sink, i, j);
+                            testN(pool, source, sink, 0, i, j);
+                            testN(pool, source, sink, i, 0, j);
+                            testN(pool, source, sink, i, j, 0);
+                            for (int k=BOUNDARY-2; k<=BOUNDARY+2; k++) {
+                                testN(pool, source, sink, i, j, k);
+                                testN(pool, source, sink, 0, i, j, k);
+                                testN(pool, source, sink, i, 0, j, k);
+                                testN(pool, source, sink, i, j, 0, k);
+                                testN(pool, source, sink, i, j, k, 0);
+                            }
+                        }
+                    }
+
+                    // Exercise write(BufferBuffer[]) on random sizes
+                    // (assumes IOV_MAX >= 8)
+                    for (int i=0; i<20; i++) {
+                        int n = rand.nextInt(9);
+                        int[] sizes = new int[n];
+                        for (int j=0; j<n; j++) {
+                            sizes[j] = rand.nextInt(1024*1024);
+                        }
+                        testN(pool, source, sink, sizes);
                     }
                 }
             }