server: kill queries on connection closed

server/handler.go

@@ -8,10 +8,10 @@ import (
 	"sync"
 	"time"
 
-	errors "gopkg.in/src-d/go-errors.v1"
 	sqle "github.com/src-d/go-mysql-server"
 	"github.com/src-d/go-mysql-server/auth"
 	"github.com/src-d/go-mysql-server/sql"
+	errors "gopkg.in/src-d/go-errors.v1"
 
 	"github.com/sirupsen/logrus"
 	"vitess.io/vitess/go/mysql"
@@ -62,6 +62,9 @@ func (h *Handler) ConnectionClosed(c *mysql.Conn) {
 	delete(h.c, c.ConnectionID)
 	h.mu.Unlock()
 
+	// If connection was closed, kill only its associated queries.
+	h.e.Catalog.ProcessList.KillOnlyQueries(c.ConnectionID)
+
 	if err := h.e.Catalog.UnlockTables(nil, c.ConnectionID); err != nil {
 		logrus.Errorf("unable to unlock tables on session close: %s", err)
 	}

sql/processlist.go

@@ -156,7 +156,7 @@ func (pl *ProcessList) AddProgressItem(pid uint64, name string, total int64) {
 	}
 }
 
-// Kill terminates all queries for a given connection
+// Kill terminates all queries for a given connection id.
 func (pl *ProcessList) Kill(connID uint32) {
 	pl.mu.Lock()
 	defer pl.mu.Unlock()
@@ -170,6 +170,21 @@ func (pl *ProcessList) Kill(connID uint32) {
 	}
 }
 
+// KillOnlyQueries kills all queries, but not index creation queries, for a
+// given connection id.
+func (pl *ProcessList) KillOnlyQueries(connID uint32) {
+	pl.mu.Lock()
+	defer pl.mu.Unlock()
+
+	for pid, proc := range pl.procs {
+		if proc.Connection == connID && proc.Type == QueryProcess {
+			logrus.Infof("kill query: pid %d", pid)
+			proc.Done()
+			delete(pl.procs, pid)
+		}
+	}
+}
+
 // Done removes the finished process with the given pid from the process list.
 // If the process does not exist, it will do nothing.
 func (pl *ProcessList) Done(pid uint64) {