session: fix packet handling race condition

server/session/session.go

@@ -313,15 +313,23 @@ func (s *Session) handlePackets() {
 		_ = s.Close()
 	}()
 	for {
-		pk, err := s.conn.ReadPacket()
-		if err != nil {
-			return
-		}
-		if err := s.handlePacket(pk); err != nil {
-			// An error occurred during the handling of a packet. Print the error and stop handling any more
-			// packets.
-			s.log.Debugf("failed processing packet from %v (%v): %v\n", s.conn.RemoteAddr(), s.c.Name(), err)
-			return
+		s.c.World().Wait()
+
+		for {
+			pk, err := s.conn.ReadPacket()
+			if err != nil {
+				return
+			}
+			if pk == nil {
+				break
+			}
+
+			if err := s.handlePacket(pk); err != nil {
+				// An error occurred during the handling of a packet. Print the error and stop handling any more
+				// packets.
+				s.log.Debugf("failed processing packet from %v (%v): %v\n", s.conn.RemoteAddr(), s.c.Name(), err)
+				return
+			}
 		}
 	}
 }

server/world/conf.go

@@ -4,6 +4,7 @@ import (
 	"github.com/df-mc/dragonfly/server/block/cube"
 	"github.com/sirupsen/logrus"
 	"math/rand"
+	"sync"
 	"time"
 )
 
@@ -80,6 +81,7 @@ func (conf Config) New() *World {
 		conf:             conf,
 		ra:               conf.Dim.Range(),
 		set:              s,
+		cond:             *sync.NewCond(&sync.Mutex{}),
 	}
 	w.weather, w.ticker = weather{w: w}, ticker{w: w}
 	var h Handler = NopHandler{}

server/world/tick.go

@@ -23,6 +23,8 @@ func (t ticker) tickLoop() {
 	for {
 		select {
 		case <-tc.C:
+			t.w.cond.Broadcast()
+
 			t.tick()
 		case <-t.w.closing:
 			// World is being closed: Stop ticking and get rid of a task.

server/world/world.go

@@ -46,6 +46,8 @@ type World struct {
 	closing chan struct{}
 	running sync.WaitGroup
 
+	cond sync.Cond
+
 	chunkMu sync.Mutex
 	// chunks holds a cache of chunks currently loaded. These chunks are cleared from this map after some time
 	// of not being used.
@@ -953,6 +955,16 @@ func (w *World) ScheduleBlockUpdate(pos cube.Pos, delay time.Duration) {
 	w.scheduledUpdates[pos] = t + delay.Nanoseconds()/int64(time.Second/20)
 }
 
+// Wait waits for broadcast from w.cond, used for packet processing.
+func (w *World) Wait() {
+	if w == nil {
+		return
+	}
+	w.cond.L.Lock()
+	w.cond.Wait()
+	w.cond.L.Unlock()
+}
+
 // doBlockUpdatesAround schedules block updates directly around and on the position passed.
 func (w *World) doBlockUpdatesAround(pos cube.Pos) {
 	if w == nil || pos.OutOfBounds(w.Range()) {