grendervill/AeroToss
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

AeroToss: оптимизация скорости передачи

Browse Source 
- Raw streams для данных (без DataInputStream/DataOutputStream overhead)
- DataInputStream/DataOutputStream только для протокольного хендшейка
- Буфер 1MB для чтения/записи
- SHA-256 вычисляется после передачи (не на каждый чанк)
- Прогресс обновляется каждую 1 сек (вместо 500мс)
- Socket: TCP_NODELAY=false, buffers=1MB, timeout=120с
- BufferedStream 4KB для протокольных данных
This commit is contained in:
grendervill
2026-08-20 15:40:07 +03:00
parent f8b73312ba
commit 37ed084e92
2 changed files with 118 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src/androidMain/kotlin/com/aerotoss/transfer/AndroidTransferManager.kt
+89 -133
View File
@@ -21,17 +21,16 @@ import java.util.concurrent.Semaphore
import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicBoolean
@OptIn(ExperimentalCoroutinesApi::class) @OptIn(ExperimentalCoroutinesApi::class)
class AndroidFileTransferManager(private val context: Context) : TransferManager { class AndroidFileTransferManager(private val context: Context) : TransferManager {
private val _incomingTransfers = MutableStateFlow<List<TransferProgress>>(emptyList()) private val _incomingTransfers = MutableStateFlow<List<TransferProgress>>(emptyList())
override val incomingTransfers: Flow<TransferProgress> = _incomingTransfers.mapLatest { list -> override val incomingTransfers: Flow<TransferProgress> = _incomingTransfers.flatMapLatest { list ->
list.lastOrNull() flow { list.forEach { emit(it) } }
}.filterNotNull() }
private val _outgoingTransfers = MutableStateFlow<List<TransferProgress>>(emptyList()) private val _outgoingTransfers = MutableStateFlow<List<TransferProgress>>(emptyList())
override val outgoingTransfers: Flow<TransferProgress> = _outgoingTransfers.mapLatest { list -> override val outgoingTransfers: Flow<TransferProgress> = _outgoingTransfers.flatMapLatest { list ->
list.lastOrNull() flow { list.forEach { emit(it) } }
}.filterNotNull() }
private var serverSocket: ServerSocket? = null private var serverSocket: ServerSocket? = null
private var serverThread: Thread? = null private var serverThread: Thread? = null
@@ -40,6 +39,7 @@ class AndroidFileTransferManager(private val context: Context) : TransferManager
private val activeSockets = ConcurrentHashMap<String, Socket>() private val activeSockets = ConcurrentHashMap<String, Socket>()
private val running = AtomicBoolean(false) private val running = AtomicBoolean(false)
private val connectionLimiter = Semaphore(10) private val connectionLimiter = Semaphore(10)
private val json = Json { ignoreUnknownKeys = true; encodeDefaults = true }
private val downloadsDir: File = run { private val downloadsDir: File = run {
val dir = context.getExternalFilesDir(Environment.DIRECTORY_DOWNLOADS) val dir = context.getExternalFilesDir(Environment.DIRECTORY_DOWNLOADS)
@@ -51,11 +51,9 @@ class AndroidFileTransferManager(private val context: Context) : TransferManager
fun startServer(port: Int = 0): Int { fun startServer(port: Int = 0): Int {
if (!running.compareAndSet(false, true)) return serverSocket?.localPort ?: 0 if (!running.compareAndSet(false, true)) return serverSocket?.localPort ?: 0
try { try {
val socket = ServerSocket(port, 50, InetAddress.getByName("0.0.0.0")) val socket = ServerSocket(port, 50, InetAddress.getByName("0.0.0.0"))
serverSocket = socket serverSocket = socket
serverThread = Thread { serverThread = Thread {
while (running.get() && !socket.isClosed) { while (running.get() && !socket.isClosed) {
try { try {
@@ -65,12 +63,7 @@ class AndroidFileTransferManager(private val context: Context) : TransferManager
if (running.get()) break if (running.get()) break
} }
} }
}.apply { }.apply { isDaemon = true; name = "aerotoss-server"; start() }
isDaemon = true
name = "aerotoss-server"
start()
}
return socket.localPort return socket.localPort
} catch (e: Exception) { } catch (e: Exception) {
e.printStackTrace() e.printStackTrace()
@@ -92,35 +85,31 @@ class AndroidFileTransferManager(private val context: Context) : TransferManager
var requestFileName: String = "" var requestFileName: String = ""
try { try {
socket.use { sock -> socket.use { sock ->
sock.soTimeout = 60_000 sock.soTimeout = 120_000
val input = DataInputStream(BufferedInputStream(sock.getInputStream(), 262144)) val rawIn = sock.getInputStream()
val output = DataOutputStream(BufferedOutputStream(sock.getOutputStream(), 262144)) val rawOut = sock.getOutputStream()
val din = DataInputStream(BufferedInputStream(rawIn, 4096))
val dout = DataOutputStream(BufferedOutputStream(rawOut, 4096))
val requestJson = input.readUTF() val requestJson = din.readUTF()
val request = try { val request = try { Json.decodeFromString<TransferRequest>(requestJson) } catch (_: Exception) { return@withContext }
Json.decodeFromString<TransferRequest>(requestJson)
} catch (_: Exception) {
return@withContext
}
requestFileName = request.fileName requestFileName = request.fileName
if (request.fileSize <= 0 || request.fileSize > FileUtils.MAX_FILE_SIZE) { if (request.fileSize < 0 || request.fileSize > FileUtils.MAX_FILE_SIZE) {
output.writeBoolean(false) dout.writeBoolean(false)
return@withContext return@withContext
} }
val progress = TransferProgress( val progress = TransferProgress(
id = progressId, id = progressId, request = request,
request = request, bytesTransferred = 0, totalBytes = request.fileSize,
bytesTransferred = 0,
totalBytes = request.fileSize,
state = TransferState.PENDING state = TransferState.PENDING
) )
updateIncoming(progress) updateIncoming(progress)
output.writeUTF(Json.encodeToString(TransferRequest.serializer(), request)) dout.writeUTF(Json.encodeToString(TransferRequest.serializer(), request))
val accepted = input.readBoolean() val accepted = din.readBoolean()
if (!accepted) { if (!accepted) {
updateIncoming(progress.copy(state = TransferState.CANCELLED)) updateIncoming(progress.copy(state = TransferState.CANCELLED))
return@withContext return@withContext
@@ -128,57 +117,53 @@ class AndroidFileTransferManager(private val context: Context) : TransferManager
val file = FileUtils.resolveUniqueFile(downloadsDir, request.fileName) val file = FileUtils.resolveUniqueFile(downloadsDir, request.fileName)
if (!FileUtils.validateFilePath(file, downloadsDir)) { if (!FileUtils.validateFilePath(file, downloadsDir)) {
output.writeBoolean(false) dout.writeBoolean(false)
return@withContext return@withContext
} }
actualFile = file actualFile = file
val sha256 = MessageDigest.getInstance("SHA-256")
var bytesWritten = 0L
updateIncoming(progress.copy(state = TransferState.TRANSFERRING)) updateIncoming(progress.copy(state = TransferState.TRANSFERRING))
var bytesWritten = 0L
val buffer = ByteArray(1048576)
file.outputStream().use { fos -> file.outputStream().use { fos ->
val buffer = ByteArray(262144)
var remaining = request.fileSize var remaining = request.fileSize
while (remaining > 0) { while (remaining > 0) {
val toRead = minOf(buffer.size.toLong(), remaining).toInt() val toRead = minOf(buffer.size.toLong(), remaining).toInt()
val read = input.read(buffer, 0, toRead) val read = rawIn.read(buffer, 0, toRead)
if (read == -1) break if (read == -1) break
fos.write(buffer, 0, read) fos.write(buffer, 0, read)
sha256.update(buffer, 0, read)
bytesWritten += read bytesWritten += read
remaining -= read remaining -= read
updateIncomingById(progressId, TransferProgress( }
id = progressId, }
request = request,
bytesTransferred = bytesWritten, val sha256 = MessageDigest.getInstance("SHA-256")
totalBytes = request.fileSize, file.inputStream().use { fis ->
state = TransferState.TRANSFERRING val buf = ByteArray(1048576)
)) while (true) {
val read = fis.read(buf)
if (read == -1) break
sha256.update(buf, 0, read)
} }
} }
val checksum = "sha256:${sha256.digest().joinToString("") { "%02x".format(it) }}" val checksum = "sha256:${sha256.digest().joinToString("") { "%02x".format(it) }}"
output.writeUTF(checksum) dout.writeUTF(checksum)
output.writeLong(bytesWritten) dout.writeLong(bytesWritten)
if (bytesWritten == request.fileSize) { if (bytesWritten == request.fileSize) {
updateIncomingById(progressId, TransferProgress( updateIncomingById(progressId, TransferProgress(
id = progressId, id = progressId, request = request,
request = request, bytesTransferred = bytesWritten, totalBytes = request.fileSize,
bytesTransferred = bytesWritten,
totalBytes = request.fileSize,
state = TransferState.COMPLETED state = TransferState.COMPLETED
)) ))
} else { } else {
FileUtils.deleteIfExists(file) FileUtils.deleteIfExists(file)
updateIncomingById(progressId, TransferProgress( updateIncomingById(progressId, TransferProgress(
id = progressId, id = progressId, request = request,
request = request, bytesTransferred = bytesWritten, totalBytes = request.fileSize,
bytesTransferred = bytesWritten, state = TransferState.FAILED, error = "Incomplete transfer"
totalBytes = request.fileSize,
state = TransferState.FAILED,
error = "Incomplete transfer: expected ${request.fileSize}, got $bytesWritten"
)) ))
} }
} }
@@ -186,12 +171,8 @@ class AndroidFileTransferManager(private val context: Context) : TransferManager
actualFile?.let { FileUtils.deleteIfExists(it) } actualFile?.let { FileUtils.deleteIfExists(it) }
val current = _incomingTransfers.value.find { it.id == progressId } val current = _incomingTransfers.value.find { it.id == progressId }
if (current != null && current.state != TransferState.COMPLETED && if (current != null && current.state != TransferState.COMPLETED &&
current.state != TransferState.FAILED && current.state != TransferState.CANCELLED current.state != TransferState.FAILED && current.state != TransferState.CANCELLED) {
) { updateIncomingById(progressId, current.copy(state = TransferState.FAILED, error = "Transfer failed"))
updateIncomingById(progressId, current.copy(
state = TransferState.FAILED,
error = "Transfer failed"
))
} }
} finally { } finally {
connectionLimiter.release() connectionLimiter.release()
@@ -200,24 +181,16 @@ class AndroidFileTransferManager(private val context: Context) : TransferManager
} }
override suspend fun sendFile(file: File, targetHost: String, targetPort: Int): Flow<TransferProgress> { override suspend fun sendFile(file: File, targetHost: String, targetPort: Int): Flow<TransferProgress> {
require(file.exists() && file.isFile) { "File does not exist or is not a regular file" } require(file.exists() && file.isFile)
require(file.length() <= FileUtils.MAX_FILE_SIZE) { "File exceeds maximum size" } require(file.length() <= FileUtils.MAX_FILE_SIZE)
val requestId = UUID.randomUUID().toString() val requestId = UUID.randomUUID().toString()
val request = TransferRequest( val request = TransferRequest(
fileName = file.name, fileName = file.name, fileSize = file.length(),
fileSize = file.length(), mimeType = "application/octet-stream", senderId = "", senderName = getDeviceName()
mimeType = "application/octet-stream",
senderId = "",
senderName = getDeviceName()
) )
val initialProgress = TransferProgress( val initialProgress = TransferProgress(
id = requestId, id = requestId, request = request,
request = request, bytesTransferred = 0, totalBytes = file.length(), state = TransferState.PENDING
bytesTransferred = 0,
totalBytes = file.length(),
state = TransferState.PENDING
) )
updateOutgoing(initialProgress) updateOutgoing(initialProgress)
@@ -230,44 +203,39 @@ class AndroidFileTransferManager(private val context: Context) : TransferManager
socket.connect(java.net.InetSocketAddress(targetHost, targetPort), 5000) socket.connect(java.net.InetSocketAddress(targetHost, targetPort), 5000)
activeSockets[requestId] = socket activeSockets[requestId] = socket
socket.use { sock -> socket.use { sock ->
sock.soTimeout = 60_000 sock.soTimeout = 120_000
val input = DataInputStream(BufferedInputStream(sock.getInputStream(), 262144)) val rawOut = sock.getOutputStream()
val output = DataOutputStream(BufferedOutputStream(sock.getOutputStream(), 262144)) val rawIn = sock.getInputStream()
val dout = DataOutputStream(BufferedOutputStream(rawOut, 4096))
val din = DataInputStream(BufferedInputStream(rawIn, 4096))
output.writeUTF(Json.encodeToString(TransferRequest.serializer(), request)) dout.writeUTF(Json.encodeToString(TransferRequest.serializer(), request))
val serverResponseJson = din.readUTF()
try { Json.decodeFromString<TransferRequest>(serverResponseJson) } catch (_: Exception) {}
dout.writeBoolean(true)
dout.flush()
val serverResponseJson = input.readUTF() updateOutgoingById(requestId, initialProgress.copy(
val serverRequest = try { state = TransferState.TRANSFERRING,
Json.decodeFromString<TransferRequest>(serverResponseJson) lastUpdateTimeMs = System.currentTimeMillis(), lastBytesTransferred = 0
} catch (_: Exception) { ))
request
}
output.writeBoolean(true)
updateOutgoingById(requestId, initialProgress.copy(state = TransferState.TRANSFERRING))
val sha256 = MessageDigest.getInstance("SHA-256")
var bytesSent = 0L var bytesSent = 0L
var lastProgressTime = System.currentTimeMillis() var lastProgressTime = System.currentTimeMillis()
var lastProgressBytes = 0L var lastProgressBytes = 0L
val buffer = ByteArray(262144) val buffer = ByteArray(1048576)
file.inputStream().use { fis -> file.inputStream().use { fis ->
while (true) { while (true) {
val read = fis.read(buffer) val read = fis.read(buffer)
if (read == -1) break if (read == -1) break
output.write(buffer, 0, read) rawOut.write(buffer, 0, read)
sha256.update(buffer, 0, read)
bytesSent += read bytesSent += read
val now = System.currentTimeMillis() val now = System.currentTimeMillis()
if (now - lastProgressTime >= 500) { if (now - lastProgressTime >= 1000) {
updateOutgoingById(requestId, TransferProgress( updateOutgoingById(requestId, TransferProgress(
id = requestId, id = requestId, request = request,
request = request, bytesTransferred = bytesSent, totalBytes = file.length(),
bytesTransferred = bytesSent, lastUpdateTimeMs = now, lastBytesTransferred = lastProgressBytes,
totalBytes = file.length(),
lastUpdateTimeMs = now,
lastBytesTransferred = lastProgressBytes,
state = TransferState.TRANSFERRING state = TransferState.TRANSFERRING
)) ))
lastProgressTime = now lastProgressTime = now
@@ -275,42 +243,37 @@ class AndroidFileTransferManager(private val context: Context) : TransferManager
} }
} }
} }
output.flush() rawOut.flush()
val serverChecksum = input.readUTF()
val bytesReceived = input.readLong()
val sha256 = MessageDigest.getInstance("SHA-256")
file.inputStream().use { fis ->
val buf = ByteArray(1048576)
while (true) { val read = fis.read(buf); if (read == -1) break; sha256.update(buf, 0, read) }
}
val localChecksum = "sha256:${sha256.digest().joinToString("") { "%02x".format(it) }}" val localChecksum = "sha256:${sha256.digest().joinToString("") { "%02x".format(it) }}"
val serverChecksum = din.readUTF()
val bytesReceived = din.readLong()
if (bytesReceived == file.length() && serverChecksum == localChecksum) { if (bytesReceived == file.length() && serverChecksum == localChecksum) {
updateOutgoingById(requestId, TransferProgress( updateOutgoingById(requestId, TransferProgress(
id = requestId, id = requestId, request = request,
request = request, bytesTransferred = file.length(), totalBytes = file.length(),
bytesTransferred = file.length(),
totalBytes = file.length(),
state = TransferState.COMPLETED state = TransferState.COMPLETED
)) ))
} else { } else {
updateOutgoingById(requestId, TransferProgress( updateOutgoingById(requestId, TransferProgress(
id = requestId, id = requestId, request = request,
request = request, bytesTransferred = bytesSent, totalBytes = file.length(),
bytesTransferred = bytesSent, state = TransferState.FAILED, error = "Checksum mismatch"
totalBytes = file.length(),
state = TransferState.FAILED,
error = "Checksum mismatch or incomplete transfer"
)) ))
} }
} }
} catch (_: Exception) { } catch (_: Exception) {
updateOutgoingById(requestId, initialProgress.copy( updateOutgoingById(requestId, initialProgress.copy(state = TransferState.FAILED, error = "Transfer failed"))
state = TransferState.FAILED,
error = "Transfer failed"
))
} finally { } finally {
activeJobs.remove(requestId) activeJobs.remove(requestId)
} }
} }
activeJobs[requestId] = job activeJobs[requestId] = job
return outgoingTransfers.filter { it.id == requestId } return outgoingTransfers.filter { it.id == requestId }
} }
@@ -329,32 +292,25 @@ class AndroidFileTransferManager(private val context: Context) : TransferManager
running.set(false) running.set(false)
activeJobs.values.forEach { it.cancel() } activeJobs.values.forEach { it.cancel() }
activeJobs.clear() activeJobs.clear()
activeSockets.clear()
scope.cancel() scope.cancel()
try { serverSocket?.close() } catch (_: Exception) {} try { serverSocket?.close() } catch (_: Exception) {}
serverThread?.interrupt() serverThread?.interrupt()
} }
private fun updateIncoming(progress: TransferProgress) { private fun updateIncoming(progress: TransferProgress) {
_incomingTransfers.update { list -> _incomingTransfers.update { list -> list.filter { it.id != progress.id }.plus(progress).takeLast(50) }
list.filter { it.id != progress.id }.plus(progress).takeLast(50)
}
} }
private fun updateIncomingById(id: String, progress: TransferProgress) { private fun updateIncomingById(id: String, progress: TransferProgress) {
_incomingTransfers.update { list -> _incomingTransfers.update { list -> list.filter { it.id != id }.plus(progress).takeLast(50) }
list.filter { it.id != id }.plus(progress).takeLast(50)
}
} }
private fun updateOutgoing(progress: TransferProgress) { private fun updateOutgoing(progress: TransferProgress) {
_outgoingTransfers.update { list -> _outgoingTransfers.update { list -> list.filter { it.id != progress.id }.plus(progress).takeLast(50) }
list.filter { it.id != progress.id }.plus(progress).takeLast(50)
}
} }
private fun updateOutgoingById(id: String, progress: TransferProgress) { private fun updateOutgoingById(id: String, progress: TransferProgress) {
_outgoingTransfers.update { list -> _outgoingTransfers.update { list -> list.filter { it.id != id }.plus(progress).takeLast(50) }
list.filter { it.id != id }.plus(progress)
}
} }
} }
app/src/desktopMain/kotlin/com/aerotoss/transfer/DesktopTransferManager.kt
+29 -19
View File
@@ -228,19 +228,19 @@ class DesktopFileTransferManager : TransferManager {
socket.connect(java.net.InetSocketAddress(targetHost, targetPort), 5000) socket.connect(java.net.InetSocketAddress(targetHost, targetPort), 5000)
activeSockets[requestId] = socket activeSockets[requestId] = socket
socket.use { sock -> socket.use { sock ->
sock.soTimeout = 60_000 sock.soTimeout = 120_000
val input = DataInputStream(BufferedInputStream(sock.getInputStream(), 262144)) val rawOut = sock.getOutputStream()
val output = DataOutputStream(BufferedOutputStream(sock.getOutputStream(), 262144)) val rawIn = sock.getInputStream()
output.writeUTF(Json.encodeToString(TransferRequest.serializer(), request)) // Protocol handshake with DataOutputStream
val dout = DataOutputStream(BufferedOutputStream(rawOut, 4096))
val din = DataInputStream(BufferedInputStream(rawIn, 4096))
val serverResponseJson = input.readUTF() dout.writeUTF(Json.encodeToString(TransferRequest.serializer(), request))
try { val serverResponseJson = din.readUTF()
Json.decodeFromString<TransferRequest>(serverResponseJson) try { Json.decodeFromString<TransferRequest>(serverResponseJson) } catch (_: Exception) {}
} catch (_: Exception) { dout.writeBoolean(true)
} dout.flush()
output.writeBoolean(true)
stateFlow.value = stateFlow.value.copy( stateFlow.value = stateFlow.value.copy(
state = TransferState.TRANSFERRING, state = TransferState.TRANSFERRING,
@@ -249,20 +249,19 @@ class DesktopFileTransferManager : TransferManager {
) )
updateOutgoing(stateFlow.value) updateOutgoing(stateFlow.value)
val sha256 = MessageDigest.getInstance("SHA-256") // File data transfer with raw streams (maximum speed)
var bytesSent = 0L var bytesSent = 0L
var lastProgressTime = System.currentTimeMillis() var lastProgressTime = System.currentTimeMillis()
var lastProgressBytes = 0L var lastProgressBytes = 0L
val buffer = ByteArray(262144) val buffer = ByteArray(1048576)
file.inputStream().use { fis -> file.inputStream().use { fis ->
while (true) { while (true) {
val read = fis.read(buffer) val read = fis.read(buffer)
if (read == -1) break if (read == -1) break
output.write(buffer, 0, read) rawOut.write(buffer, 0, read)
sha256.update(buffer, 0, read)
bytesSent += read bytesSent += read
val now = System.currentTimeMillis() val now = System.currentTimeMillis()
if (now - lastProgressTime >= 500) { if (now - lastProgressTime >= 1000) {
stateFlow.value = stateFlow.value.copy( stateFlow.value = stateFlow.value.copy(
bytesTransferred = bytesSent, bytesTransferred = bytesSent,
lastUpdateTimeMs = now, lastUpdateTimeMs = now,
@@ -275,10 +274,21 @@ class DesktopFileTransferManager : TransferManager {
} }
} }
} }
output.flush() rawOut.flush()
val serverChecksum = input.readUTF() // Compute checksum after transfer
val bytesReceived = input.readLong() val sha256 = MessageDigest.getInstance("SHA-256")
file.inputStream().use { fis ->
val buf = ByteArray(1048576)
while (true) {
val read = fis.read(buf)
if (read == -1) break
sha256.update(buf, 0, read)
}
}
val checksum = "sha256:${sha256.digest().joinToString("") { "%02x".format(it) }}"
val serverChecksum = din.readUTF()
val bytesReceived = din.readLong()
val localChecksum = "sha256:${sha256.digest().joinToString("") { "%02x".format(it) }}" val localChecksum = "sha256:${sha256.digest().joinToString("") { "%02x".format(it) }}"