MultiThread fucking works

2026-05-16 01:25:21 +03:00
parent 19c0b2842f
commit 655a0d50d6
3 changed files with 2274 additions and 88 deletions
@@ -3,15 +3,17 @@ package collection
 import models.HumanBeing
 import mu.KotlinLogging
 import java.time.LocalDate
+import java.util.Collections
 import java.util.TreeSet
 import java.util.UUID

 private val logger = KotlinLogging.logger {}

-
 class CollectionManager {

-    private val collection: TreeSet<HumanBeing> = TreeSet()
+    private val collection: MutableSet<HumanBeing> =
+        Collections.synchronizedSortedSet(TreeSet())
+
    val initDate: LocalDate = LocalDate.now()

    fun add(humanBeing: HumanBeing): Boolean {
@@ -20,11 +22,13 @@ class CollectionManager {
    }

    fun update(id: UUID, updated: HumanBeing): Boolean {
-        val old = getById(id) ?: return false
-        collection.remove(old)
-        val replaced = updated.copy(id = old.id, creationDate = old.creationDate)
-        logger.info { "[UPDATE] Element updated ${old.id}" }
-        return collection.add(replaced)
+        synchronized(collection) {
+            val old = getById(id) ?: return false
+            collection.remove(old)
+            val replaced = updated.copy(id = old.id, creationDate = old.creationDate)
+            logger.info { "[UPDATE] Element updated ${old.id}" }
+            return collection.add(replaced)
+        }
    }

    fun removeById(id: UUID): Boolean {
@@ -38,47 +42,57 @@ class CollectionManager {
        logger.info { "[CLEAR] Collection cleared" }
    }

-    fun getAll(): TreeSet<HumanBeing> = TreeSet(collection)
+    fun getAll(): TreeSet<HumanBeing> {
+        synchronized(collection) {
+            return TreeSet(collection)
+        }
+    }

-    fun getById(id: UUID): HumanBeing? =
-        collection.stream()
-            .filter { it.id == id }
-            .findFirst()
-            .orElse(null)
+    fun getById(id: UUID): HumanBeing? {
+        synchronized(collection) {
+            return collection.firstOrNull { it.id == id }
+        }
+    }

    fun size(): Int = collection.size

    fun isEmpty(): Boolean = collection.isEmpty()

-    fun getMax(): HumanBeing? =
-        collection.stream()
-            .max(Comparator.naturalOrder())
-            .orElse(null)
+    fun getMax(): HumanBeing? {
+        synchronized(collection) {
+            return collection.maxWithOrNull(Comparator.naturalOrder())
+        }
+    }

-    fun getMin(): HumanBeing? =
-        collection.stream()
-            .min(Comparator.naturalOrder())
-            .orElse(null)
+    fun getMin(): HumanBeing? {
+        synchronized(collection) {
+            return collection.minWithOrNull(Comparator.naturalOrder())
+        }
+    }

-    fun sumOfMinutesOfWaiting(): Double =
-        collection.stream()
-            .mapToDouble { it.minutesOfWaiting.toDouble() }
-            .sum()
+    fun sumOfMinutesOfWaiting(): Double {
+        synchronized(collection) {
+            return collection.sumOf { it.minutesOfWaiting.toDouble() }
+        }
+    }

-    fun minByName(): HumanBeing? =
-        collection.stream()
-            .min(Comparator.comparing { it.name })
-            .orElse(null)
+    fun minByName(): HumanBeing? {
+        synchronized(collection) {
+            return collection.minByOrNull { it.name }
+        }
+    }

-    fun getMinutesOfWaitingDescending(): List<Float> =
-        collection.stream()
-            .map { it.minutesOfWaiting }
-            .sorted(Comparator.reverseOrder())
-            .toList()
+    fun getMinutesOfWaitingDescending(): List<Float> {
+        synchronized(collection) {
+            return collection.map { it.minutesOfWaiting }.sortedDescending()
+        }
+    }

    fun loadFromFile(items: List<HumanBeing>) {
-        collection.clear()
-        items.stream().forEach { collection.add(it) }
+        synchronized(collection) {
+            collection.clear()
+            collection.addAll(items)
+        }
        logger.info { "[LOAD] loaded ${items.size} elements" }
    }

@@ -7,86 +7,132 @@ import runner.CommandInvoker
 import java.net.DatagramPacket
 import java.net.DatagramSocket
 import java.net.InetAddress
+import java.util.concurrent.Executors
+import java.util.concurrent.ForkJoinPool
+import java.util.concurrent.RecursiveAction
+import java.util.concurrent.RecursiveTask

 private val logger = KotlinLogging.logger {}

-/**
- * UDP-сервер на основе DatagramSocket (датаграммы).
- *
- * Однопоточный: receive → deserialize → execute → send.
- *
- * @property port порт прослушивания
- * @property invoker инвокер серверных команд
- */
 class NetworkManager(
    private val port: Int,
    private val invoker: CommandInvoker,
    private val repository: Repository
 ) {
    private val socket = DatagramSocket(port)
+    private val readerPool = Executors.newFixedThreadPool(4)
+    private val processingPool = ForkJoinPool(Runtime.getRuntime().availableProcessors())
+    private val senderPool = ForkJoinPool(Runtime.getRuntime().availableProcessors())

    fun start() {
-        logger.info{"Server started on port $port. Waiting for connection..."}
-        val buf = ByteArray(65507)
+        logger.info { "Server started on port $port. Waiting for connection..." }
+        repeat(4) {
+            readerPool.submit(::readLoop)
+        }
+        try {
+            Thread.currentThread().join()
+        } catch (_: InterruptedException) {}
+    }

+    private fun readLoop() {
+        val buf = ByteArray(65507)
        while (!socket.isClosed) {
            try {
-                val packet = DatagramPacket(buf, buf.size)
+                val packet = DatagramPacket(buf.copyOf(), buf.size)
                socket.receive(packet)
+
+                val address = packet.address
+                val clientPort = packet.port
                val json = String(packet.data, 0, packet.length, Charsets.UTF_8)

-                val authRequest = try {
-                    AppJson.decodeFromString<AuthRequest>(json)
-                } catch (_: SerializationException) {
-                    val request = try {
-                        AppJson.decodeFromString<Request>(json)
-                    } catch (e: SerializationException) {
-                        logger.warn(e) { "Invalid request: ${e.message}" }
-                        sendResponse(Response(false, "Некорректный формат запроса"), packet.address, packet.port)
-                        continue
-                    }
-                    logger.info{"'${request.commandName}' from ${packet.address}:${packet.port}."}
-
-                    val response = invoker.execute(request)
-                    sendResponse(response, packet.address, packet.port)
-                    continue
-                }
-
-                val token : String?
-                val id: String?
-                if (authRequest.type == AuthType.LOGIN) {
-                    val res = repository.login(authRequest.login, authRequest.password)
-                    token = res?.first; id = res?.second
-                } else {
-                    val res = repository.register(authRequest.login, authRequest.password)
-                    token = res?.first; id = res?.second
-                }
-
-                if (token != null) {
-                    sendResponse(AuthResponse(true, "success", token, id), packet.address, packet.port)
-                } else {
-                    sendResponse(AuthResponse(false, "login failed", token, id), packet.address, packet.port)
-                }
+                processingPool.submit(ProcessRequestTask(json, address, clientPort))

            } catch (e: Exception) {
-                if (!socket.isClosed)
-                    logger.warn(e) { "Error: ${e.message}" }
+                if (!socket.isClosed) logger.warn(e) { "Read error: ${e.message}" }
+                else break
            }
        }
    }

-    private fun sendResponse(response: Response, address: InetAddress, port: Int) {
-        val bytes = AppJson.encodeToString(Response.serializer(), response).toByteArray(Charsets.UTF_8)
-        socket.send(DatagramPacket(bytes, bytes.size, address, port))
+    inner class ProcessRequestTask(
+        private val json: String,
+        private val address: InetAddress,
+        private val clientPort: Int
+    ) : RecursiveTask<Unit>() {
+
+        override fun compute() {
+            val response: Any = try {
+                val authRequest = try {
+                    AppJson.decodeFromString<AuthRequest>(json)
+                } catch (_: SerializationException) { null }
+
+                if (authRequest != null) {
+                    handleAuth(authRequest)
+                } else {
+                    val request = try {
+                        AppJson.decodeFromString<Request>(json)
+                    } catch (e: SerializationException) {
+                        logger.warn(e) { "Invalid request: ${e.message}" }
+                        senderPool.submit(
+                            SendResponseTask(
+                                Response(false, "Некорректный формат запроса"),
+                                address, clientPort
+                            )
+                        )
+                        return
+                    }
+                    logger.info { "'${request.commandName}' from $address:$clientPort." }
+                    invoker.execute(request)
+                }
+            } catch (e: Exception) {
+                Response(false, "Внутренняя ошибка сервера: ${e.message}")
+            }
+
+            senderPool.submit(SendResponseTask(response, address, clientPort))
+        }
+
+        private fun handleAuth(authRequest: AuthRequest): AuthResponse {
+            val res = if (authRequest.type == AuthType.LOGIN)
+                repository.login(authRequest.login, authRequest.password)
+            else
+                repository.register(authRequest.login, authRequest.password)
+
+            return if (res != null)
+                AuthResponse(true, "success", res.first, res.second)
+            else
+                AuthResponse(false, "login failed", null, null)
+        }
    }

-    private fun sendResponse(response: AuthResponse, address: InetAddress, port: Int) {
-        val bytes = AppJson.encodeToString(AuthResponse.serializer(), response).toByteArray(Charsets.UTF_8)
-        socket.send(DatagramPacket(bytes, bytes.size, address, port))
+    inner class SendResponseTask(
+        private val response: Any,
+        private val address: InetAddress,
+        private val clientPort: Int
+    ) : RecursiveAction() {
+
+        override fun compute() {
+            try {
+                val bytes = when (response) {
+                    is Response     -> AppJson.encodeToString(Response.serializer(), response)
+                    is AuthResponse -> AppJson.encodeToString(AuthResponse.serializer(), response)
+                    else -> return
+                }.toByteArray(Charsets.UTF_8)
+
+                val responsePacket = DatagramPacket(bytes, bytes.size, address, clientPort)
+                synchronized(socket) {
+                    socket.send(responsePacket)
+                }
+            } catch (e: Exception) {
+                logger.warn(e) { "Send error: ${e.message}" }
+            }
+        }
    }

    fun stop() {
-        logger.info{ "Shutting down..." }
+        logger.info { "Shutting down..." }
        socket.close()
+        readerPool.shutdownNow()
+        processingPool.shutdown()
+        senderPool.shutdown()
    }
 }