From f95b67f93d77aae1a18561e8d7d41f0ab294e801 Mon Sep 17 00:00:00 2001 From: andrei-mihnea-cerbu Date: Wed, 13 Nov 2024 16:25:25 +0200 Subject: [PATCH] network chunk v19 --- .../socket_communicator_base.ts | 39 +------------------ .../tcp_client_communicator.ts | 16 ++++---- .../tcp_server_communicator.ts | 16 ++++---- 3 files changed, 17 insertions(+), 54 deletions(-) diff --git a/User/src/network/socket_communicator/socket_communicator_base.ts b/User/src/network/socket_communicator/socket_communicator_base.ts index f5dc5c1..f749193 100644 --- a/User/src/network/socket_communicator/socket_communicator_base.ts +++ b/User/src/network/socket_communicator/socket_communicator_base.ts @@ -1,6 +1,5 @@ import { ParsedMessage } from '../message_handler'; import { OperationHandler } from '../operations_base/operation_handler'; -import ping from "ping"; import { constants, createCipheriv, @@ -16,16 +15,16 @@ export abstract class SocketCommunicatorBase { protected readonly port: number; protected readonly operationHandler: OperationHandler; protected handlerResult: ParsedMessage | null; - protected networkSpeed: number | null = null; protected chunkBuffers: { [messageId: string]: string[] }; - protected readonly EOP = ''; protected privateKey: string | null; protected publicKey: string | null; protected aesKey: Buffer | null; protected aesIv: Buffer | null; + protected readonly EOP = ''; + protected readonly CHUNK_SIZE = 1024; private incompleteChunkBuffer: string = ''; protected constructor(ip: string, port: number, operationHandler: OperationHandler) { @@ -113,40 +112,6 @@ export abstract class SocketCommunicatorBase { ).toString('base64'); } - protected async scanNetworkLatency(): Promise { - const targetIp = this.ip; // Use the IP from the superclass - - try { - const response = await ping.promise.probe(targetIp); - - if (!response.alive || response.time === "unknown") { - console.warn(`Ping failed to reach ${targetIp}. Using default network speed.`); - return 200; // Default latency in ms if ping fails - } - - return response.time; // Latency in ms from ping response - } catch (error: any) { - console.error(`Ping error: ${error.message}. Using default network speed.`); - return 200; // Default latency in ms if an error occurs - } - } - - // Calculate optimal chunk size based on network latency, with fallback if necessary - protected async calculateOptimalChunkSize(messageLength: number): Promise { - const latency = await this.scanNetworkLatency(); - this.networkSpeed = latency > 0 ? 1000 / latency : 1; // Speed in bytes/ms based on latency - - // Calculate initial chunk size based on latency (bounded between 512 and 1024 bytes) - let chunkSize = Math.min(Math.max(512, Math.floor(5000 / this.networkSpeed)), 1024); - - // Adjust chunk size for base64 alignment (multiple of 4) - while (messageLength % chunkSize !== 0 && chunkSize > 0) { - chunkSize -= 4; - } - - return chunkSize || 1024; // Fallback to 1024 if alignment adjustment results in 0 - } - async handleIncomingChunk(data: Buffer): Promise { // Append incoming data to the incomplete buffer this.incompleteChunkBuffer += data.toString(); diff --git a/User/src/network/socket_communicator/tcp_client_communicator.ts b/User/src/network/socket_communicator/tcp_client_communicator.ts index 51c956a..11824b8 100644 --- a/User/src/network/socket_communicator/tcp_client_communicator.ts +++ b/User/src/network/socket_communicator/tcp_client_communicator.ts @@ -58,29 +58,29 @@ export class TcpClientCommunicator extends SocketCommunicatorBase { } async sendMessage(operationCode: string, metaInfo?: { [key: string]: any }, fileContent?: Buffer): Promise { - if (!this.networkSpeed) { - this.networkSpeed = await this.scanNetworkLatency(); - } - const message = MessageHandler.formatMessage(operationCode, metaInfo, fileContent); const outgoingMessage = this.encryptWithAes(message); // Calculate optimal chunk size based on network latency - const optimalChunkSize = await this.calculateOptimalChunkSize(outgoingMessage.length); - const totalChunks = Math.ceil(outgoingMessage.length / optimalChunkSize); + const totalChunks = Math.ceil(outgoingMessage.length / this.CHUNK_SIZE); + const messageId = Date.now().toString(); // Send each chunk with a delay between them for (let i = 0; i < totalChunks; i++) { - const chunk = outgoingMessage.slice(i * optimalChunkSize, (i + 1) * optimalChunkSize); + const chunk = outgoingMessage.slice(i * this.CHUNK_SIZE, (i + 1) * this.CHUNK_SIZE); const chunkHeader = JSON.stringify({ messageId, sequenceNumber: i + 1, totalChunks, }); const chunkWithHeader = `${chunkHeader}|${chunk}${this.EOP}`; - while(!this.socket.write(chunkWithHeader)); + + if (!this.socket.write(chunkWithHeader)) { + // Wait for the 'drain' event before writing the next chunk + await new Promise((resolve) => this.socket.once('drain', resolve)); + } } } diff --git a/User/src/network/socket_communicator/tcp_server_communicator.ts b/User/src/network/socket_communicator/tcp_server_communicator.ts index 4bbfb79..aa1779b 100644 --- a/User/src/network/socket_communicator/tcp_server_communicator.ts +++ b/User/src/network/socket_communicator/tcp_server_communicator.ts @@ -38,10 +38,6 @@ export class TcpServerCommunicator extends SocketCommunicatorBase { } async sendMessage(operationCode: string, metaInfo?: { [key: string]: any }, fileContent?: Buffer): Promise { - if (!this.networkSpeed) { - this.networkSpeed = await this.scanNetworkLatency(); - } - const message = MessageHandler.formatMessage(operationCode, metaInfo, fileContent); let outgoingMessage: string; @@ -56,21 +52,23 @@ export class TcpServerCommunicator extends SocketCommunicatorBase { outgoingMessage = this.encryptWithAes(message); } - // Calculate optimal chunk size based on network latency - const optimalChunkSize = await this.calculateOptimalChunkSize(outgoingMessage.length); - const totalChunks = Math.ceil(outgoingMessage.length / optimalChunkSize); + const totalChunks = Math.ceil(outgoingMessage.length / this.CHUNK_SIZE); const messageId = Date.now().toString(); // Send each chunk with a delay between them for (let i = 0; i < totalChunks; i++) { - const chunk = outgoingMessage.slice(i * optimalChunkSize, (i + 1) * optimalChunkSize); + const chunk = outgoingMessage.slice(i * this.CHUNK_SIZE, (i + 1) * this.CHUNK_SIZE); const chunkHeader = JSON.stringify({ messageId, sequenceNumber: i + 1, totalChunks, }); const chunkWithHeader = `${chunkHeader}|${chunk}${this.EOP}`; - while(!this.socket.write(chunkWithHeader)); + + if (!this.socket.write(chunkWithHeader)) { + // Wait for the 'drain' event before writing the next chunk + await new Promise((resolve) => this.socket.once('drain', resolve)); + } } } }