无效Session不需要向服务器端发送SESSION_CLOSE命令

lealone-client/src/main/java/org/lealone/client/ClientSession.java

@@ -125,7 +125,7 @@ public Session connect(boolean first) {
                 Session[] sessions = new ClientSession[size];
                 for (int i = 0; i < size; i++) {
                     // 如果首次连接的节点就是复制节点之一，则复用它
-                    if (!isInvalid()) {
+                    if (isValid()) {
                         HostAndPort hostAndPort = new HostAndPort(servers[i]);
                         if (hostAndPort.inetSocketAddress.equals(this.inetSocketAddress)) {
                             sessions[i] = this;
@@ -259,7 +259,8 @@ private Transfer initTransfer(ConnectionInfo ci, HostAndPort hostAndPort) throws
         sessionId = getNextId();
         transfer = asyncConnection.createTransfer(this);
         asyncConnection.writeInitPacket(this, sessionId, transfer, ci);
-        asyncConnection.addSession(sessionId, this);
+        if (isValid())
+            asyncConnection.addSession(sessionId, this);
         return transfer;
     }
 
@@ -367,8 +368,11 @@ public void close() {
         synchronized (this) {
             try {
                 traceOperation("SESSION_CLOSE", 0);
-                transfer.writeRequestHeader(sessionId, Session.SESSION_CLOSE).flush();
-                asyncConnection.removeSession(sessionId);
+                // 只有当前Session有效时服务器端才持有对应的session
+                if (isValid()) {
+                    transfer.writeRequestHeader(sessionId, Session.SESSION_CLOSE).flush();
+                    asyncConnection.removeSession(sessionId);
+                }
 
                 synchronized (ClientSession.class) {
                     if (asyncConnection.isEmpty()) {

lealone-common/src/main/java/org/lealone/db/Session.java

@@ -194,6 +194,8 @@ public interface Session extends Closeable, Transaction.Participant {
 
     boolean isInvalid();
 
+    boolean isValid();
+
     void setTargetEndpoints(String targetEndpoints);
 
     String getTargetEndpoints();

lealone-common/src/main/java/org/lealone/db/SessionBase.java

@@ -102,6 +102,11 @@ public boolean isInvalid() {
         return invalid;
     }
 
+    @Override
+    public boolean isValid() {
+        return !invalid;
+    }
+
     @Override
     public void setTargetEndpoints(String targetEndpoints) {
         this.targetEndpoints = targetEndpoints;

lealone-db/src/main/java/org/lealone/db/DatabaseEngine.java

@@ -88,7 +88,7 @@ public ServerSession createSession(ConnectionInfo ci) {
                         // ignore
                     }
                 }
-                if (session.isInvalid()) {
+                if (session.isInvalid()) { // 无效session，不需要进行后续的操作
                     return session;
                 }
 

lealone-net/src/main/java/org/lealone/net/AsyncConnection.java

@@ -64,47 +64,23 @@ static class SessionInfo {
         final Session session;
         final CommandHandler commandHandler;
         final ConcurrentLinkedQueue<PreparedCommand> preparedCommandQueue;
+        final long commandHandlerSequence;
 
         SessionInfo(String hostAndPort, int sessionId, Session session, CommandHandler commandHandler) {
             this.sessionId = sessionId;
             this.hostAndPort = hostAndPort;
             this.session = session;
             this.commandHandler = commandHandler;
             this.preparedCommandQueue = new ConcurrentLinkedQueue<>();
+
+            // SessionInfo的hostAndPort和sessionId字段不足以区别它自身，所以用commandHandlerSequence
+            commandHandlerSequence = commandHandler.getNextSequence();
         }
 
         @Override
         public String toString() {
             return "SessionInfo [hostAndPort=" + hostAndPort + ", sessionId=" + sessionId + "]";
         }
-
-        @Override
-        public int hashCode() {
-            final int prime = 31;
-            int result = 1;
-            result = prime * result + ((hostAndPort == null) ? 0 : hostAndPort.hashCode());
-            result = prime * result + sessionId;
-            return result;
-        }
-
-        @Override
-        public boolean equals(Object obj) {
-            if (this == obj)
-                return true;
-            if (obj == null)
-                return false;
-            if (getClass() != obj.getClass())
-                return false;
-            SessionInfo other = (SessionInfo) obj;
-            if (hostAndPort == null) {
-                if (other.hostAndPort != null)
-                    return false;
-            } else if (!hostAndPort.equals(other.hostAndPort))
-                return false;
-            if (sessionId != other.sessionId)
-                return false;
-            return true;
-        }
     }
 
     private static final Logger logger = LoggerFactory.getLogger(AsyncConnection.class);
@@ -221,7 +197,7 @@ private void readInitPacket(Transfer transfer, int sessionId) {
             String userName = transfer.readString();
             userName = StringUtils.toUpperEnglish(userName);
             Session session = createSession(transfer, originalURL, dbName, userName);
-            if (!session.isInvalid()) {
+            if (session.isValid()) {
                 CommandHandler commandHandler = CommandHandler.getNextCommandHandler();
                 sessions.put(sessionId, session);
                 SessionInfo sessionInfo = new SessionInfo(hostAndPort, sessionId, session, commandHandler);
@@ -772,16 +748,16 @@ protected void processRequest(Transfer transfer, int id, int operation) throws I
             int sessionId = transfer.readInt();
             Session session = getSession(sessionId);
             session.commit(transfer.readString());
-            writeResponseHeader(transfer, session, id);
-            transfer.flush();
+            // writeResponseHeader(transfer, session, id); //不需要发回响应
+            // transfer.flush();
             break;
         }
         case Session.COMMAND_DISTRIBUTED_TRANSACTION_ROLLBACK: {
             int sessionId = transfer.readInt();
             Session session = getSession(sessionId);
             session.rollback();
-            writeResponseHeader(transfer, session, id);
-            transfer.flush();
+            // writeResponseHeader(transfer, session, id); //不需要发回响应
+            // transfer.flush();
             break;
         }
         case Session.COMMAND_DISTRIBUTED_TRANSACTION_ADD_SAVEPOINT:
@@ -793,8 +769,8 @@ protected void processRequest(Transfer transfer, int id, int operation) throws I
                 session.addSavepoint(name);
             else
                 session.rollbackToSavepoint(name);
-            writeResponseHeader(transfer, session, id);
-            transfer.flush();
+            // writeResponseHeader(transfer, session, id); //不需要发回响应
+            // transfer.flush();
             break;
         }
         case Session.COMMAND_DISTRIBUTED_TRANSACTION_VALIDATE: {
@@ -892,9 +868,13 @@ protected void processRequest(Transfer transfer, int id, int operation) throws I
         }
         case Session.SESSION_CLOSE: {
             SessionInfo si = sessionInfoMap.remove(id);
-            si.commandHandler.removeSession(si);
-            Session session = sessions.remove(id);
-            closeSession(session);
+            if (si != null) {
+                si.commandHandler.removeSession(si);
+                Session session = sessions.remove(id);
+                closeSession(session);
+            } else {
+                logger.warn("SessionInfo is null, may be a bug! sessionId = " + id);
+            }
             break;
         }
         case Session.SESSION_CANCEL_STATEMENT: {

lealone-net/src/main/java/org/lealone/net/CommandHandler.java

@@ -23,6 +23,7 @@
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 
 import org.lealone.db.SessionStatus;
 import org.lealone.net.AsyncConnection.SessionInfo;
@@ -74,17 +75,23 @@ public static void removeConnection(AsyncConnection c) {
         c.close();
     }
 
-    private final ConcurrentHashMap<SessionInfo, SessionInfo> sessionInfoMap = new ConcurrentHashMap<>();
+    private final AtomicLong sequence = new AtomicLong(0);
+    private final ConcurrentHashMap<Long, SessionInfo> sessionInfoMap = new ConcurrentHashMap<>();
     private final Semaphore haveWork = new Semaphore(1);
     private boolean stop;
     private int nested;
 
+    long getNextSequence() {
+        return sequence.incrementAndGet();
+    }
+
     void addSession(SessionInfo sessionInfo) {
-        sessionInfoMap.put(sessionInfo, sessionInfo);
+        // SessionInfo的hostAndPort和sessionId字段不足以区别它自身，所以用commandHandlerSequence
+        sessionInfoMap.put(sessionInfo.commandHandlerSequence, sessionInfo);
     }
 
     void removeSession(SessionInfo sessionInfo) {
-        sessionInfoMap.remove(sessionInfo);
+        sessionInfoMap.remove(sessionInfo.commandHandlerSequence);
     }
 
     public CommandHandler(int id) {