如果多个协议包同时出现时有可能后面一个是不完整的

lealone · Mar 19, 2016 · f770db6 · f770db6
1 parent 67d6bc4
commit f770db6
Show file tree

Hide file tree

Showing 3 changed files with 18 additions and 14 deletions.
diff --git a/lealone-mvcc/src/main/java/org/lealone/mvcc/log/LogSyncService.java b/lealone-mvcc/src/main/java/org/lealone/mvcc/log/LogSyncService.java
@@ -18,7 +18,7 @@
 package org.lealone.mvcc.log;
 
 import java.util.ArrayList;
-import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.LinkedTransferQueue;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 
@@ -30,7 +30,7 @@ public abstract class LogSyncService extends Thread {
     protected final Semaphore haveWork = new Semaphore(1);
     protected final WaitQueue syncComplete = new WaitQueue();
 
-    protected final LinkedBlockingQueue<MVCCTransaction> transactions = new LinkedBlockingQueue<>();
+    protected final LinkedTransferQueue<MVCCTransaction> transactions = new LinkedTransferQueue<>();
 
     protected long syncIntervalMillis;
     protected volatile long lastSyncedAt = System.currentTimeMillis();

diff --git a/lealone-net/src/main/java/org/lealone/net/AsyncConnection.java b/lealone-net/src/main/java/org/lealone/net/AsyncConnection.java
@@ -504,8 +504,10 @@ private void processResponse(Transfer transfer, int id) throws IOException {
         }
 
         AsyncCallback<?> ac = callbackMap.remove(id);
-        if (ac == null)
+        if (ac == null) {
+            logger.warn("Async callback is null, may be a bug! id = " + id);
             return;
+        }
         if (e != null)
             ac.setDbException(e);
         else
@@ -983,10 +985,6 @@ public void handle(Buffer buffer) {
         }
         int pos = 0;
         while (true) {
-            if (length < 4) {
-                tmpBuffer = buffer;
-                break;
-            }
             int packetLength = buffer.getInt(pos);
             if (length - 4 == packetLength) {
                 if (pos == 0) {
@@ -1004,7 +1002,15 @@ public void handle(Buffer buffer) {
 
                 pos = pos + packetLength + 4;
                 length = length - (packetLength + 4);
-                continue;
+
+                // 有可能剩下的不够4个字节了
+                if (length < 4) {
+                    tmpBuffer = Buffer.buffer();
+                    tmpBuffer.appendBuffer(buffer, pos, length);
+                    break;
+                } else {
+                    continue;
+                }
             } else {
                 tmpBuffer = Buffer.buffer();
                 tmpBuffer.appendBuffer(buffer, pos, length);

diff --git a/lealone-test/src/test/java/org/lealone/test/misc/CRUDExample.java b/lealone-test/src/test/java/org/lealone/test/misc/CRUDExample.java
@@ -192,13 +192,10 @@ static void benchmark() throws Exception {
         Statement stmt = conn.createStatement();
         stmt.executeUpdate("DROP TABLE IF EXISTS test");
         stmt.executeUpdate("CREATE TABLE IF NOT EXISTS test (f1 int primary key, f2 long)");
-        // // stmt.executeUpdate("set MULTI_THREADED 1");
+        stmt.close();
-        //
-        // // stmt.close();
-        // // conn.close();
 
-        int threadsCount = 1;// Runtime.getRuntime().availableProcessors() * 4;// 10;
+        int threadsCount = 10;// Runtime.getRuntime().availableProcessors() * 4;// 10;
-        int loop = 100;
+        int loop = 500;
         latch = new CountDownLatch(threadsCount);
 
         MyThread[] threads = new MyThread[threadsCount];
@@ -211,6 +208,7 @@ static void benchmark() throws Exception {
         }
 
         latch.await();
+        conn.close();
 
         long write_sum = 0;
         for (int i = 0; i < threadsCount; i++) {