Handling errors in Dask distributed

I have a data processing server that will receive Dask arrays (I send them through writing them to hdf5 files). The server reads the file, performs some computations on the Dask array using the distributed framework, and then writes the results to a new hdf5 file and send this back to the client. This works in so far as there are no errors during the computation. I can do this repeatedly without issue.

I wanted to install error handling into the server, so that if there is some sort of error during the computation, that the client receives the error message. This also works, with a caveat. The problem I have is that if there is an error, and subsequently we receive a new job to perform a fresh computation, then the computation is performed, but we can no longer write to the hdf5 file: it gives me a "ValueError: Can only serialize read-only h5py files".

Here is an outline of the code - it's not the full code with all the functions but it should be enough to get the idea. In particular, the function write_dataset() just uses the Dask .to_hdf() method and this is where the error happens. I believe this is somehow related to the fact that the processed_output variable is stored within Dask distributed already, so when we redo the computation, although it works, it is not rewritten and the permissions have changed. I don't know how to fix this issue, so any thoughts on how to clear the variables after a failed Dask distributed computation are appreciated.

```

import logging
import socket
import os
import h5py
import io
import json

error_file_name = 'error.log'
log_file_path = 'console.log'
file_name = 'temp_file.h5'

if __name__ == '__main__':
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind((HOST, PORT))

while True:

s.listen()
conn, address = s.accept()

#Receive the HDF5 file in chunks
chunks = []
while True:
chunk = conn.recv(1024)
if not chunk:
break
chunks.append(chunk)

#if there is already a processed output variable, delete it.
try:
del processed_output
except:
pass

#Read the hdf5 file
with h5py.File(io.BytesIO(b''.join(chunks)), 'r') as f:
f.flush()
#read the dataset
h5_dataset = f['MyDataGroup/sid_data/sid_data']

print('Received dataset {}'.format(h5_dataset))

try:
processed_output = do_some_processing(h5_dataset)
except ValueError as e:
logging.error(f"ValueError: {e}")

except Exception as e:
logging.error(f"Unexpected error: {e}")

#If we received an error message...
if 'processed_output' not in locals():
try:
os.remove(error_file_name)
except:
pass

#Read the log
print('in error handling section')
with open(log_file_path, 'r') as f:
log_contents = f.read()
log_lines = log_contents.split('\n')
f.flush()

json_data = json.dumps(log_lines, indent=4)

with open(error_file_name, 'a') as f:
f.write(json_data)
f.flush()

with open(error_file_name, 'rb') as f:
f.flush()
img = f.read()
print('Failed; sending the error back to client')

#if not...
else:
try:
os.remove(file_name)
except:
pass

with h5py.File(file_name, 'a') as h5_f:
data_group = h5_f.create_group('MyDataGroup')
for ind,dsets in enumerate(processed_output):
print(ind, dsets)
write_dataset(dsets,data_group, main_data_name = 'processed_data_' + str(ind))
h5_f.flush()
img = h5_f.id.get_file_image()
print('now sending the data back to client')

#Now send it back

conn, address = s.accept()
conn.sendall(img)

#Close off the connections
conn.close()
```