logos-blockchain/logos-blockchain-specs
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-blockchain-specs.git synced 2026-04-29 22:13:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add helper methods for dispersal errors

Browse Source
This commit is contained in:
Gusto 2024-07-11 13:43:16 +03:00
parent a72dab131a
commit 46495f04a5
No known key found for this signature in database
4 changed files with 40 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
da/executor/executor.py
View File

@ -22,7 +22,7 @@ class Executor:
except asyncio.CancelledError:
break
except Exception as e:
print(f"Error during message sending: {e}")
print(f"Executor: Error during message sending: {e}")
async def connect(self):
try:
@ -30,18 +30,18 @@ class Executor:
conn_id = len(self.connections)
transport = Transport(conn_id, reader, writer, self._handle)
self.connections.append(transport)
print(f"Connected to {self.addr}:{self.port}, ID: {conn_id}")
print(f"Executor: Connected to {self.addr}:{self.port}, ID: {conn_id}")
asyncio.create_task(transport.read_and_process())
except Exception as e:
print(f"Failed to connect or lost connection: {e}")
print(f"Executor: Failed to connect or lost connection: {e}")
async def _handle(self, conn_id, writer, message):
if message.HasField('dispersal_res'):
print(f"Received DispersalRes: blob_id={message.dispersal_res.blob_id}")
print(f"Executor: Received DispersalRes: blob_id={message.dispersal_res.blob_id}")
elif message.HasField('sample_res'):
print(f"Received SampleRes: blob_id={message.sample_res.blob_id}")
print(f"Executor: Received SampleRes: blob_id={message.sample_res.blob_id}")
else:
print("Received unknown message type")
print("Executor: Received unknown message type")
async def run(self):
await asyncio.gather(*(self.connect() for _ in range(self.col_num)))

2
da/executor/mock_network.py
View File

@ -24,8 +24,6 @@ def main():
# TODO: Add args parser.
app = App()
# asyncio.run(app.run_nodes(10000, 4096))
# asyncio.run(app.run_executor('localhost', 8888, 1))
asyncio.run(app.run_node_with_executor(1))
if __name__ == '__main__':

11
da/executor/node.py
View File

@ -20,17 +20,20 @@ class Node:
server = await asyncio.start_server(
self._on_conn, self.addr, self.port
)
print(f"Server started at {self.addr}:{self.port}")
print(f"Node: Server started at {self.addr}:{self.port}")
async with server:
await server.serve_forever()
async def _handle(self, conn_id, writer, message):
if message.HasField('dispersal_req'):
print(f"Received DispersalRes: blob_id={message.dispersal_req.blob.blob_id}")
blob_id = message.dispersal_req.blob.blob_id
print(f"Node: Received DispersalRes: blob_id={blob_id}")
# Imitate succesful verification.
writer.write(proto.new_dispersal_res_success_msg(blob_id))
elif message.HasField('sample_req'):
print(f"Received SampleRes: blob_id={message.sample_req.blob_id}")
print(f"Node: Received SampleRes: blob_id={message.sample_req.blob_id}")
else:
print("Received unknown message type")
print("Node: Received unknown message type")
async def run(self):
await self.listen()

48
da/executor/proto.py
View File

@ -20,17 +20,23 @@ def new_dispersal_req_msg(blob_id, data):
dispersal_message = dispersal_pb2.DispersalMessage(dispersal_req=dispersal_req)
return pack_message(dispersal_message)
def new_dispersal_res_msg(blob_id=None, error_type=None, error_description=None):
if blob_id is not None:
blob_id_msg = dispersal_pb2.BlobId(blob_id=blob_id)
dispersal_res = dispersal_pb2.DispersalRes(blob_id=blob_id_msg)
elif error_type is not None and error_description is not None:
error = dispersal_pb2.Error(description=error_description)
dispersal_err = dispersal_pb2.DispersalErr()
setattr(dispersal_err, error_type, error)
dispersal_res = dispersal_pb2.DispersalRes(err=dispersal_err)
else:
raise ValueError("Either blob_id or error_type and error_description must be provided")
def new_dispersal_res_success_msg(blob_id):
blob_id_msg = dispersal_pb2.BlobId(blob_id=blob_id)
dispersal_res = dispersal_pb2.DispersalRes(blob_id=blob_id_msg)
dispersal_message = dispersal_pb2.DispersalMessage(dispersal_res=dispersal_res)
return pack_message(dispersal_message)
def new_dispersal_res_chunk_size_error_msg(description):
error = dispersal_pb2.Error(description=description)
dispersal_err = dispersal_pb2.DispersalErr(chunk_size_err=error)
dispersal_res = dispersal_pb2.DispersalRes(err=dispersal_err)
dispersal_message = dispersal_pb2.DispersalMessage(dispersal_res=dispersal_res)
return pack_message(dispersal_message)
def new_dispersal_res_verification_error_msg(description):
error = dispersal_pb2.Error(description=description)
dispersal_err = dispersal_pb2.DispersalErr(verification_err=error)
dispersal_res = dispersal_pb2.DispersalRes(err=dispersal_err)
dispersal_message = dispersal_pb2.DispersalMessage(dispersal_res=dispersal_res)
return pack_message(dispersal_message)
@ -40,16 +46,16 @@ def new_sample_req_msg(blob_id):
dispersal_message = dispersal_pb2.DispersalMessage(sample_req=sample_req)
return pack_message(dispersal_message)
def new_sample_res_msg(blob_id=None, data=None, error_description=None):
if blob_id is not None and data is not None:
blob = dispersal_pb2.Blob(blob_id=blob_id, data=data)
sample_res = dispersal_pb2.SampleRes(blob=blob)
elif error_description is not None:
error = dispersal_pb2.Error(description=error_description)
sample_err = dispersal_pb2.SampleErr(not_found=error)
sample_res = dispersal_pb2.SampleRes(err=sample_err)
else:
raise ValueError("Either blob_id and data or error_description must be provided")
def new_sample_res_success_msg(blob_id, data):
blob = dispersal_pb2.Blob(blob_id=blob_id, data=data)
sample_res = dispersal_pb2.SampleRes(blob=blob)
dispersal_message = dispersal_pb2.DispersalMessage(sample_res=sample_res)
return pack_message(dispersal_message)
def new_sample_res_not_found_error_msg(description):
error = dispersal_pb2.Error(description=description)
sample_err = dispersal_pb2.SampleErr(not_found=error)
sample_res = dispersal_pb2.SampleRes(err=sample_err)
dispersal_message = dispersal_pb2.DispersalMessage(sample_res=sample_res)
return pack_message(dispersal_message)