Fixing payload misalignment/corruption issue #14
Conversation
… process header and payload together
| @@ -60,6 +60,8 @@ def __init__(self, logger=None, ip_address_family="IPv4"): | |||
| else: | |||
| self.logger.error("Wrong ip_address_family - {}, Supported {}".format(ip_address_family, IP_ADDRESS_FAMILY)) | |||
| raise ConnectionError("Socket initialization failed! ") | |||
| # set socket to send data as a burst rather than keeping data in the buffer | |||
| msg = "{}" | ||
| time_started = datetime.now() | ||
|
|
||
| try: | ||
| msg_len_in_bytes = b'' | ||
| received_msg_len_size = 0 |
There was a problem hiding this comment.
unused variable. Remove it
| if msg_len: | ||
| # first receive message length from payload | ||
| msg_len_in_bytes = self.socket.recv(MESSAGE_LENGTH) | ||
| msg_len = int(msg_len_in_bytes) |
There was a problem hiding this comment.
Make sure the received data is equal to MESSAGE_LENGTH
| if msg_body == b'': | ||
| raise RuntimeError("ClientSocket: Empty message for message length -{}".format(msg_len)) | ||
|
|
||
| if msg_body: | ||
| elif msg_body: | ||
| if len(msg_body) == msg_len: |
There was a problem hiding this comment.
why do we need to do len(msg_body) when the received_data_size is available on the top.
| return json_data | ||
| else: | ||
| return msg | ||
| if format.upper() == "JSON": |
There was a problem hiding this comment.
This goes wrong with the code landing on line 171. Check it
| except socket.timeout as e: | ||
| self.logger.error("ClientSocket: Timeout ({} seconds) from recv function".format((datetime.now() - time_started).seconds)) | ||
| # return empty status and let the next iteration or worker process the status left on the buffer | ||
| return json.loads(msg) |
There was a problem hiding this comment.
since the msg is filled up before and got any exception, you are sending wrong data to the caller. Check again
| @@ -358,8 +357,8 @@ def recv_json(self, format="JSON", timeout=RECV_TIMEOUT): | |||
| json_data = {} | |||
| try: | |||
| json_data = json.loads(msg) | |||
| except json.JSONDecodeError as e: | |||
| raise json.JSONDecodeError("ClientSocket: Bad JSON data - {}".format(e)) | |||
| # except json.JSONDecodeError as e: | |||
There was a problem hiding this comment.
what is the reason for commenting out
| @@ -325,6 +321,9 @@ def recv_json(self, format="JSON", timeout=RECV_TIMEOUT): | |||
| msg_len = int(msg_len_in_bytes.decode('utf8')) | |||
| except socket.timeout as e: | |||
| raise e | |||
| except BlockingIOError as e: | |||
There was a problem hiding this comment.
Here also we need to add the above logic where the header and payload are received together similar to clientsocket
|
Verified latest changes on client and server side with 80GB of data. |
|
Kudos, SonarCloud Quality Gate passed! |
grandsuri
deleted the
13-data-mover-payload-misalignment-corruption-on-larger-data-migrations
branch
Verified and tested on 4 node dist. cluster with 80GB of data