Add various locks in osuToken object

events/startSpectatingEvent.py

@@ -8,6 +8,10 @@ def handle(userToken, packetData):
 		# Start spectating packet
 		packetData = clientPackets.startSpectating(packetData)
 
+		# If the user id is less than 0, treat this as a stop spectating packet
+		if packetData["userID"] < 0:
+			userToken.stopSpectating()
+
 		# Get host token
 		targetToken = glob.tokens.getTokenFromUserID(packetData["userID"])
 		if targetToken is None:

handlers/mainHandler.pyx

@@ -93,7 +93,7 @@ class handler(requestsManager.asyncRequestHandler):
 
 				# Token exists, get its object and lock it
 				userToken = glob.tokens.tokens[requestTokenString]
-				userToken.lock.acquire()
+				userToken.processingLock.acquire()
 
 				# Keep reading packets until everything has been read
 				while pos < len(requestData):
@@ -205,8 +205,8 @@ class handler(requestsManager.asyncRequestHandler):
 				if userToken is not None:
 					# Update ping time for timeout
 					userToken.updatePingTime()
-					# Release token lock
+					# Release processing lock
-					userToken.lock.release()
+					userToken.processingLock.release()
 					# Delete token if kicked
 					if userToken.kicked:
 						glob.tokens.deleteToken(userToken)

objects/osuToken.py

@@ -39,8 +39,6 @@ class token:
 		self.loginTime = int(time.time())
 		self.pingTime = self.loginTime
 		self.timeOffset = timeOffset
-		self.lock = threading.Lock()
-		self.bufferLock = threading.Lock()
 		self.streams = []
 		self.tournament = tournament
 		self.messagesBuffer = []
@@ -87,6 +85,15 @@ class token:
 		else:
 			self.token = str(uuid.uuid4())
 
+		# Locks
+		self.processingLock = threading.Lock()	# Acquired while there's an incoming packet from this user
+		self._bufferLock = threading.Lock()		# Acquired while writing to packets buffer
+		self._internalLock = threading.Lock()	# Acquired while performing internal operations on this token
+		self._chatLock = threading.Lock()		# Acquired while performing chat operations
+		self._streamsLock = threading.Lock()	# Acquired while joining/leaving streams
+		self._spectatorLock = threading.Lock()	# Acquired while starting/stopping spectating
+		self._multiplayerLock = threading.Lock()# Acquired while joining/leaving streams
+
 		# Set stats
 		self.updateCachedStats()
 
@@ -105,7 +112,7 @@ class token:
 		"""
 		try:
 			# Acquire the buffer lock
-			self.bufferLock.acquire()
+			self._bufferLock.acquire()
 
 			# Never enqueue for IRC clients or Foka
 			if self.irc or self.userID < 999:
@@ -118,15 +125,15 @@ class token:
 				log.warning("{}'s packets buffer is above 10M!! Lost some data!".format(self.username))
 		finally:
 			# Release the buffer lock
-			self.bufferLock.release()
+			self._bufferLock.release()
 
 	def resetQueue(self):
 		"""Resets the queue. Call when enqueued packets have been sent"""
 		try:
-			self.bufferLock.acquire()
+			self._bufferLock.acquire()
 			self.queue = bytes()
 		finally:
-			self.bufferLock.release()
+			self._bufferLock.release()
 
 	def joinChannel(self, channelObject):
 		"""
@@ -136,6 +143,8 @@ class token:
 		:raises: exceptions.userAlreadyInChannelException()
 				 exceptions.channelNoPermissionsException()
 		"""
+		try:
+			self._chatLock.acquire()
 			if channelObject.name in self.joinedChannels:
 				raise exceptions.userAlreadyInChannelException()
 			if channelObject.publicRead == False and self.admin == False:
@@ -143,6 +152,8 @@ class token:
 			self.joinedChannels.append(channelObject.name)
 			self.joinStream("chat/{}".format(channelObject.name))
 			self.enqueue(serverPackets.channelJoinSuccess(self.userID, channelObject.clientName))
+		finally:
+			self._chatLock.release()
 
 	def partChannel(self, channelObject):
 		"""
@@ -150,8 +161,12 @@ class token:
 
 		:param channelObject: channel object
 		"""
+		try:
+			self._chatLock.acquire()
 			self.joinedChannels.remove(channelObject.name)
 			self.leaveStream("chat/{}".format(channelObject.name))
+		finally:
+			self._chatLock.release()
 
 	def setLocation(self, latitude, longitude):
 		"""
@@ -185,9 +200,13 @@ class token:
 
 		:param host: host osuToken object
 		"""
+		try:
 			# Stop spectating old client
 			self.stopSpectating()
 
+			# Acquire token's lock
+			self._spectatorLock.acquire()
+
 			# Set new spectator host
 			self.spectating = host.token
 			self.spectatingUserID = host.userID
@@ -221,6 +240,8 @@ class token:
 
 			# Log
 			log.info("{} is spectating {}".format(self.username, host.username))
+		finally:
+			self._spectatorLock.release()
 
 	def stopSpectating(self):
 		"""
@@ -229,6 +250,10 @@ class token:
 
 		:return:
 		"""
+		try:
+			# Acquire token lock
+			self._spectatorLock.acquire()
+
 			# Remove our userID from host's spectators
 			if self.spectating is None:
 				return
@@ -266,6 +291,8 @@ class token:
 			# Set our spectating user to 0
 			self.spectating = None
 			self.spectatingUserID = 0
+		finally:
+			self._spectatorLock.release()
 
 	def updatePingTime(self):
 		"""
@@ -282,6 +309,9 @@ class token:
 		:param matchID: new match ID
 		:return:
 		"""
+		try:
+			self._multiplayerLock.acquire()
+
 			# Make sure the match exists
 			if matchID not in glob.matches.matches:
 				return
@@ -311,6 +341,8 @@ class token:
 			# Alert the user if we have just joined a tourney match
 			if match.isTourney:
 				self.enqueue(serverPackets.notification("You are now in a tournament match."))
+		finally:
+			self._multiplayerLock.release()
 
 	def leaveMatch(self):
 		"""
@@ -318,6 +350,9 @@ class token:
 
 		:return:
 		"""
+		try:
+			self._multiplayerLock.acquire()
+
 			# Make sure we are in a match
 			if self.matchID == -1:
 				return
@@ -340,6 +375,8 @@ class token:
 
 			# Set slot to free
 			match.userLeft(self)
+		finally:
+			self._multiplayerLock.release()
 
 	def kick(self, message="You have been kicked from the server. Please login again.", reason="kick"):
 		"""
@@ -350,6 +387,9 @@ class token:
 		:param reason: Kick reason, used in logs. Default: "kick"
 		:return:
 		"""
+		try:
+			self._internalLock.acquire()
+
 			# Send packet to target
 			log.info("{} has been disconnected. ({})".format(self.username, reason))
 			if message != "":
@@ -358,6 +398,8 @@ class token:
 
 			# Logout event
 			logoutEvent.handle(self, deleteToken=self.irc)
+		finally:
+			self._internalLock.release()
 
 	def silence(self, seconds = None, reason = "", author = 999):
 		"""
@@ -368,6 +410,9 @@ class token:
 		:param author: userID of who has silenced the user. Default: 999 (FokaBot)
 		:return:
 		"""
+		try:
+			self._chatLock.acquire()
+
 			if seconds is None:
 				# Get silence expire from db if needed
 				seconds = max(0, userUtils.getSilenceEnd(self.userID) - int(time.time()))
@@ -383,6 +428,8 @@ class token:
 
 			# Send silenced packet to everyone else
 			glob.streams.broadcast("main", serverPackets.userSilenced(self.userID))
+		finally:
+			self._chatLock.release()
 
 	def spamProtection(self, increaseSpamRate = True):
 		"""
@@ -391,6 +438,9 @@ class token:
 		:param increaseSpamRate: set to True if the user has sent a new message. Default: True
 		:return:
 		"""
+		try:
+			self._chatLock.acquire()
+
 			# Increase the spam rate if needed
 			if increaseSpamRate:
 				self.spamRate += 1
@@ -398,6 +448,8 @@ class token:
 			# Silence the user if needed
 			if self.spamRate > 10:
 				self.silence(1800, "Spamming (auto spam protection)")
+		finally:
+			self._chatLock.release()
 
 	def isSilenced(self):
 		"""
@@ -422,6 +474,9 @@ class token:
 
 		:return:
 		"""
+		try:
+			self._internalLock.acquire()
+
 			stats = userUtils.getUserStats(self.userID, self.gameMode)
 			log.debug(str(stats))
 			if stats is None:
@@ -433,6 +488,8 @@ class token:
 			self.totalScore = stats["totalScore"]
 			self.gameRank = stats["gameRank"]
 			self.pp = stats["pp"]
+		finally:
+			self._internalLock.release()
 
 	def checkRestricted(self):
 		"""
@@ -484,9 +541,13 @@ class token:
 		:param name: stream name
 		:return:
 		"""
+		try:
+			self._streamsLock.acquire()
 			glob.streams.join(name, token=self.token)
 			if name not in self.streams:
 				self.streams.append(name)
+		finally:
+			self._streamsLock.release()
 
 	def leaveStream(self, name):
 		"""
@@ -495,9 +556,13 @@ class token:
 		:param name: stream name
 		:return:
 		"""
+		try:
+			self._streamsLock.acquire()
 			glob.streams.leave(name, token=self.token)
 			if name in self.streams:
 				self.streams.remove(name)
+		finally:
+			self._streamsLock.release()
 
 	def leaveAllStreams(self):
 		"""