Fix masked array deserialization overflow for integer dtypes

NumPy masked arrays default to a fill value of 999999, which cannot be
represented by small integer dtypes such as uint8 or uint16.

During distributed deserialization this raises a TypeError when
reconstructing the masked array and may cause tasks to hang indefinitely.

Ensure that fill values are cast safely to the target dtype, falling back
to NumPy's default fill value when necessary.

Author: abslock128w

distributed/protocol/numpy.py

@@ -216,4 +216,12 @@ def deserialize_numpy_maskedarray(header, frames):
     if pickled_fv:
         fill_value = pickle.loads(fill_value)
 
+       # Ensure fill_value is compatible with dtype
+    if fill_value is not None:
+        try:
+            fill_value = np.array(fill_value, dtype=data.dtype).item()
+        except (OverflowError, ValueError, TypeError):
+            fill_value = np.ma.default_fill_value(data.dtype)
+
     return np.ma.masked_array(data, mask=mask, fill_value=fill_value)
+