Introduce a variety of approaches to transfer a pickable data, config

proxy/common_neon/config.py

@@ -0,0 +1,27 @@
+from abc import ABC, abstractmethod
+from typing import Optional
+import os
+
+
+class IConfig(ABC):
+
+    @abstractmethod
+    def get_solana_url(self) -> Optional[str]:
+        """Gets the predefinded solana url"""
+
+    @abstractmethod
+    def get_evm_count(self) -> Optional[int]:
+        """Gets the evm count"""
+
+
+class Config(IConfig):
+
+    def __init__(self):
+        from ..environment import read_elf_params, ELF_PARAMS
+        read_elf_params(ELF_PARAMS)
+
+    def get_solana_url(self) -> Optional[str]:
+        return os.environ.get("SOLANA_URL", "http://localhost:8899")
+
+    def get_evm_count(self) -> Optional[int]:
+        return int(os.environ.get("EVM_STEP_COUNT", 750))

proxy/common_neon/data.py

@@ -1,11 +1,19 @@
 from __future__ import annotations
 from dataclasses import dataclass
 from typing import Dict, Any
+from .eth_proto import Trx as NeonTx
 
 
 @dataclass
-class NeonTxPrecheckResult:
-    is_underpriced_tx_without_chainid: bool
+class NeonTxCfg:
+    is_without_chainid: bool
+    steps_executed: int
+
+
+@dataclass
+class MemPoolTxCfg:
+    neon_tx: NeonTx
+    neon_tx_cfg: NeonTxCfg
     emulating_result: NeonEmulatingResult
 
 

proxy/common_neon/utils/__init__.py

@@ -1,4 +1,5 @@
 from .utils import *
 
-from .pickable_data_server import PickableDataServer, PickableDataServerUser, PickableDataClient
+from .pickable_data_server import AddrPickableDataSrv, PipePickableDataSrv, PickableDataServerUser, \
+                                  AddrPickableDataClient, PipePickableDataClient
 

proxy/common_neon/utils/pickable_data_server.py

@@ -1,84 +1,152 @@
+from typing import Any, Tuple
 from abc import ABC, abstractmethod
+
 import asyncio
+from asyncio import StreamReader, StreamWriter
 import socket
 import pickle
 import struct
-from typing import Any
 from logged_groups import logged_group
 
 
 class PickableDataServerUser(ABC):
 
     @abstractmethod
-    def on_data_received(self, data: Any):
+    def on_data_received(self, data: Any) -> Any:
         """Gets neon_tx_data from the neon rpc api service worker"""
 
 
+def encode_pickable(object) -> bytes:
+    data = pickle.dumps(object)
+    len_data = struct.pack("!I", len(data))
+    return len_data + data
+
+
 @logged_group("neon.MemPool")
 class PickableDataServer(ABC):
 
-    def __init__(self, *, user: PickableDataServerUser, host: str, port: int):
+    def __init__(self, *, user: PickableDataServerUser):
         self._user = user
-        self._port = port
-        self._host = host
+        asyncio.get_event_loop().create_task(self.run_server())
+
+    @abstractmethod
+    async def run_server(self):
+        assert False
 
-    async def handle_client(self, client):
-        loop = asyncio.get_event_loop()
-        peer_name = client.getpeername()
-        self.debug(f"Got new incoming connection: {peer_name}")
+    async def handle_client(self, reader: StreamReader, writer: StreamWriter):
         while True:
             try:
-                len_packed: bytes = await loop.sock_recv(client, 4)
-                if len(len_packed) == 0:
-                    break
-                # TODO: all the data can be received by parts, handle it
-                payload_len_data = struct.unpack("!I", len_packed[:4])[0]
-                payload = await loop.sock_recv(client, payload_len_data)
-                data = pickle.loads(payload)
-                self._user.on_data_received(data)
+                data = await self._recv_pickable_data(reader)
+                result = self._user.on_data_received(data)
+                result_data = encode_pickable(result)
+                writer.write(result_data)
+                await writer.drain()
             except ConnectionResetError:
-                self.error(f"Client connection: {peer_name} - has been interrupted")
+                self.error(f"Client connection has been closed")
                 break
             except Exception as err:
-                self.error(f"Failed to receive data over: {peer_name} - err: {err}")
-                continue
-        client.close()
+                self.error(f"Failed to receive data err: {err}, type: {type(err)}")
+                break
+
+    async def _recv_pickable_data(self, reader: StreamReader):
+        len_packed: bytes = await reader.readexactly(4)
+        if len(len_packed) == 0:
+            raise ConnectionResetError()
+        payload_len_data = struct.unpack("!I", len_packed)[0]
+        payload = await reader.readexactly(payload_len_data)
+        data = pickle.loads(payload)
+
+        return data
+
+
+class AddrPickableDataSrv(PickableDataServer):
+
+    def __init__(self, *, user: PickableDataServerUser, address: Tuple[str, int]):
+        self._address = address
+        PickableDataServer.__init__(self, user=user)
 
     async def run_server(self):
-        self.info(f"Listen port: {self._port} on: {self._host}")
-        server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        server.bind((self._host, self._port))
-        server.listen(8)
-        server.setblocking(False)
+        host, port = self._address
+        self.info(f"Listen port: {port} on: {host}")
+        await asyncio.start_server(self.handle_client, host, port)
 
-        loop = asyncio.get_event_loop()
-        while True:
-            client, _ = await loop.sock_accept(server)
-            loop.create_task(self.handle_client(client))
+
+class PipePickableDataSrv(PickableDataServer):
+
+    def __init__(self, *, user: PickableDataServerUser, srv_sock: socket.socket):
+        self._srv_sock = srv_sock
+        PickableDataServer.__init__(self, user=user)
+
+    async def run_server(self):
+        print("run_server_by_conn")
+        reader, writer = await asyncio.streams.open_connection(sock=self._srv_sock)
+        print("Got reader, writer")
+        await self.handle_client(reader, writer)
 
 
 @logged_group("neon.Proxy")
 class PickableDataClient:
 
-    def __init__(self, host: str, port: int):
+    CONNECTION_TIMEOUT_SEC = 5
 
-        self.info(f"Initialize PickableDataClient connecting to: {port} at: {host}")
-        self._connection = socket.create_connection((host, port))
+    def __init__(self):
+        self._client_sock = None
 
-    def send_data(self, pickable_data: Any):
+    def _set_client_sock(self, client_sock: socket.socket):
+        self._client_sock = client_sock
+        self._client_sock.setblocking(False)
+        self._client_sock.settimeout(self.CONNECTION_TIMEOUT_SEC)
+
+    def send_data(self, pickable_object: Any):
         try:
-            payload = self._encode_pickable_data(pickable_data)
-            self._connection.send(payload)
+            payload = encode_pickable(pickable_object)
+            self._client_sock.send(payload)
+            len_packed: bytes = self._client_sock.recv(4)
+            data_len = struct.unpack("!I", len_packed)[0]
+            data = self._client_sock.recv(data_len)
+            if not data:
+                return None
+            result = pickle.loads(data)
+            return result
         except BaseException as err:
             self.error(f"Failed to send data: {err}")
             raise Exception("Failed to send pickable data")
 
-    def _encode_pickable_data(self, pickable_data: Any):
-        data = pickle.dumps(pickable_data)
-        data_len = len(data)
-        packed_len = struct.pack("!I", data_len)
-        payload = packed_len + data
-        return payload
+    async def send_data_async(self, pickable_object):
+        reader, writer = await asyncio.streams.open_connection(sock=self._client_sock)
+        try:
+            payload = encode_pickable(pickable_object)
+            writer.write(payload)
+            await writer.drain()
+            len_packed: bytes = await reader.readexactly(4)
+            if not len_packed:
+                return None
+            data_len = struct.unpack("!I", len_packed)[0]
+            data = await reader.readexactly(data_len)
+            if not data:
+                return None
+            result = pickle.loads(data)
+            return result
+        except BaseException as err:
+            self.error(f"Failed to send data: {err}")
+            raise Exception("Failed to send pickable data")
 
     def __del__(self):
-        self._connection.close()
+        self._client_sock.close()
+
+
+class PipePickableDataClient(PickableDataClient):
+
+    def __init__(self, client_sock: socket.socket):
+        PickableDataClient.__init__(self)
+        self._set_client_sock(client_sock=client_sock)
+
+
+class AddrPickableDataClient(PickableDataClient):
+
+    def __init__(self, addr: Tuple[str, int]):
+        PickableDataClient.__init__(self)
+        host, port = addr
+        client_sock = socket.create_connection((host, port))
+        self._set_client_sock(client_sock=client_sock)
+

proxy/common_neon/utils/utils.py

@@ -6,9 +6,6 @@
 
 from eth_utils import big_endian_to_int
 
-#TODO: move it out from here
-from ...environment import EVM_LOADER_ID
-
 from ..eth_proto import Trx as EthTx
 
 
@@ -129,6 +126,8 @@ def fill_block_info(self, block: SolanaBlockInfo):
             rec['blockNumber'] = hex(block.slot)
 
     def decode(self, neon_sign: str, tx: {}, ix_idx=-1) -> NeonTxResultInfo:
+        # TODO: move it out from here
+        from ...environment import EVM_LOADER_ID
         self._set_defaults()
         meta_ixs = tx['meta']['innerInstructions']
         msg = tx['transaction']['message']