Finally all tests passing

Author: awicenec

daliuge-common/dlg/utils.py

@@ -528,12 +528,12 @@ def prepareUser(DLG_ROOT=getDlgDir()):
 
 
 def serialize_data(d):
-    # return pickle.dumps(d)
+    # return dill.dumps(d)
     return b2s(base64.b64encode(dill.dumps(d)))
 
 
 def deserialize_data(d):
-    # return pickle.loads()
+    # return dill.loads(d)
     return dill.loads(base64.b64decode(d.encode("utf8")))
 
 

daliuge-engine/dlg/apps/pyfunc.py

@@ -33,7 +33,7 @@
 import pickle
 import re
 
-from typing import Callable
+from typing import Callable, Union
 import dill
 from io import StringIO
 from contextlib import redirect_stdout
@@ -68,7 +68,8 @@ def serialize_func(f):
         parts = f.split(".")
         f = getattr(importlib.import_module(".".join(parts[:-1])), parts[-1])
 
-    fser = dill.dumps(f)
+    fser = base64.b64encode(dill.dumps(f)).decode()
+    # fser = inspect.getsource(f)
     fdefaults = {"args": [], "kwargs": {}}
     adefaults = {"args": [], "kwargs": {}}
     a = inspect.getfullargspec(f)
@@ -139,32 +140,43 @@ def import_using_name(app, fname):
             logger.debug("Loaded module: %s", mod)
             return mod
 
+def import_using_code_ser(func_code: Union[str, bytes], func_name: str):
+    """
+    Import the function provided as a serialised code string.
+    """
+    try:
+        func_code = func_code if isinstance(func_code, bytes) else func_code.encode()
+        func = dill.loads(base64.b64decode(func_code))
+    except Exception as err:
+        logger.warning("Unable to deserialize func_code: %s", err)
+        raise
+    if func_name and func_name.split(".")[-1] != func.__name__:
+        raise ValueError(
+            f"Function with name '{func.__name__}' instead of '{func_name}' found!"
+        )
+    return func
 
 def import_using_code(func_code: str, func_name: str, serialized: bool = True):
     """
     Import the function provided as a code string. Plain code as well as serialized code
     is supported. If the func_name does not match the provided func_name the load will fail.
     """
-    if not serialized:
-        logger.debug(f"Trying to import code from string: {func_code}")
-        mod = pyext.RuntimeModule.from_string("mod", func_name, func_code)
+    mod = None
+    if not serialized and not isinstance(func_code, bytes):
+        logger.debug(f"Trying to import code from string: {func_code.strip()}")
+        try:
+            mod = pyext.RuntimeModule.from_string("mod", func_name, func_code.strip())
+        except Exception:
+            func = import_using_code_ser(func_code, func_name)
         logger.debug("Imported function: %s", func_name)
-        if func_name:
+        if mod and func_name:
             if hasattr(mod, func_name):
                 func = getattr(mod, func_name)
             else:
                 logger.warning("Function with name '%s' not found!", func_name)
                 raise ValueError(f"Function with name '{func_name}' not found!")
     else:
-        try:
-            func = dill.loads(func_code)
-        except Exception as err:
-            logger.warning("Unable to deserialize func_code: %s", err)
-            raise
-        if func_name and func_name.split(".")[-1] != func.__name__:
-            raise ValueError(
-                f"Function with name '{func.__name__}' instead of '{func_name}' found!"
-            )
+        func = import_using_code_ser(func_code, func_name)
     logger.debug("Imported function: %s", func_name)
     return func
 
@@ -570,16 +582,16 @@ def initialize_with_func_code(self):
                 )
             except (SyntaxError, NameError) as err:
                 logger.warning(
-                    "Problem importing code: %s. Checking whether it was serialized.",
+                    "Problem importing code: %s.",
                     err,
                 )
-                serialized = True
-        if isinstance(self.func_code, bytes) or serialized:
-            if isinstance(self.func_code, str):
-                self.func_code = base64.b64decode(self.func_code.encode("utf8"))
-            self.func = import_using_code(
-                self.func_code, self.func_name, serialized=True
-            )
+                # serialized = True
+        # if isinstance(self.func_code, bytes) or serialized:
+        #     if isinstance(self.func_code, str):
+        #         self.func_code = base64.b64decode(self.func_code.encode("utf8"))
+        #     self.func = import_using_code(
+        #         self.func_code, self.func_name, serialized=True
+        #     )
 
         self._init_fn_defaults()
         # make sure defaults are dicts
@@ -778,9 +790,9 @@ def write_results(self, result):
                 result = result_iter[i]
 
             parser = self._match_parser(o)
-            if parser is DropParser.PICKLE:
-                logger.debug(f"Writing pickeled result {type(result)} to {o}")
-                o.write(pickle.dumps(result))
+            if parser in [DropParser.PICKLE, DropParser.DILL]:
+                logger.debug(f"Writing dilled result {type(result)} to {o}")
+                o.write(dill.dumps(result))
             elif parser is DropParser.EVAL or parser is DropParser.UTF8:
                 encoded_result = repr(result).encode("utf-8")
                 o.write(encoded_result)
@@ -795,8 +807,6 @@ def write_results(self, result):
                 drop_loaders.save_npy(o, result)
             elif parser is DropParser.RAW:
                 o.write(result)
-            elif parser is DropParser.DILL:
-                o.write(dill.dumps(result))
             elif parser is DropParser.BINARY:
                 drop_loaders.save_binary(o, result)
             else:

daliuge-engine/dlg/apps/simple.py

@@ -20,6 +20,7 @@
 #    MA 02111-1307  USA
 #
 """Applications used as examples, for testing, or in simple situations"""
+import base64
 import dill
 import _pickle
 from numbers import Number
@@ -1007,9 +1008,12 @@ def initialize(self, **kwargs):
     def readData(self):
         input = self.inputs[0]
         data = pickle.loads(droputils.allDropContents(input))
+        # data = droputils.allDropContents(input)
+        # data = dill.loads(base64.b64decode(data))
 
         # make sure we always have a ndarray with at least 1dim.
         if type(data) not in (list, tuple) and not isinstance(data, (np.ndarray)):
+            logger.warning("Data type not in [list, tuple]: %s", data)
             raise TypeError
         if isinstance(data, np.ndarray) and data.ndim == 0:
             data = np.array([data])

daliuge-engine/dlg/dask_emulation.py

@@ -25,6 +25,7 @@
 
 import base64
 import contextlib
+import inspect
 import logging
 import pickle
 import socket
@@ -312,6 +313,7 @@ def __init__(self, f, nout):
         if hasattr(f, "__name__"):
             self.fname = f.__name__
         self.fcode, self.fdefaults = pyfunc.serialize_func(f)
+        # self.fcode = inspect.getsource(f)
         self.original_kwarg_names = []
         self.original_arg_names = []
         self.nout = nout
@@ -336,7 +338,10 @@ def make_dropdict(self):
             my_dropdict["func_name"] = self.fname
             my_dropdict["name"] = simple_fname
         if self.fcode is not None:
-            my_dropdict["func_code"] = utils.b2s(base64.b64encode(self.fcode))
+            logger.debug("func_code provided: %s", self.fcode)
+            my_dropdict["func_code"] = self.fcode
+            # my_dropdict["func_code"] = base64.b64encode(self.fcode)
+            # my_dropdict["func_code"] = utils.b2s(base64.b64encode(self.fcode))
         if self.fdefaults:
             # APPLICATION ARGUMENTS
             my_dropdict["func_defaults"] = self.fdefaults
@@ -435,6 +440,7 @@ def make_dropdict(self):
         )
         if not self.producer:
             my_dropdict["pydata"] = pyfunc.serialize_data(self.pydata)
+            # my_dropdict["pydata"] = self.pydata
         return my_dropdict
 
     def __repr__(self):

daliuge-engine/dlg/data/drops/data_base.py

@@ -97,7 +97,7 @@ def decrRefCount(self):
         with self._refLock:
             self._refCount -= 1
 
-    @track_current_drop
+    # @track_current_drop
     def open(self, **kwargs):
         """
         Opens the DROP for reading, and returns a "DROP descriptor"

daliuge-engine/dlg/data/drops/memory.py

@@ -20,32 +20,48 @@
 #    MA 02111-1307  USA
 #
 import base64
+import builtins
+import dill
 import io
 import json
 import os
-import pickle
 import random
 import string
 import sys
+from typing import Union
 
 from dlg.common.reproducibility.reproducibility import common_hash
 from dlg.data.drops.data_base import DataDROP, logger
 from dlg.data.io import SharedMemoryIO, MemoryIO
 
+def get_builtins()-> dict:
+    """
+    Get a tuple of buitlin types to compare pydata with.
+    """
+    builtin_types = tuple(getattr(builtins, t) for t in dir(builtins) if isinstance(getattr(builtins, t), type))
+    builtin_types = builtin_types[builtin_types.index(bool):]
+    builtin_names = [b.__name__ for b in builtin_types]
+    return dict(zip(builtin_names, builtin_types))
 
-def parse_pydata(pd_dict: dict) -> bytes:
+
+def parse_pydata(pd: Union[bytes, dict]) -> bytes:
     """
     Parse and evaluate the pydata argument to populate memory during initialization
 
-    :param pd_dict: the pydata dictionary from the graph node
+    :param pd: either the pydata dictionary from the graph node or the value directly
 
     :returns a byte encoded value
     """
+    pd_dict = pd if isinstance(pd, dict) else {"value":pd, "type":"raw"}
     pydata = pd_dict["value"]
-    logger.debug(f"pydata value provided: {pydata}, {pd_dict['type'].lower()}")
+    logger.debug("pydata value provided: '%s' with type '%s'", pydata, type(pydata))
 
     if pd_dict["type"].lower() in ["string", "str"]:
         return pydata if pydata != "None" else None
+    builtin_types = get_builtins()
+    if pd_dict["type"] != "raw" and type(pydata) in builtin_types.values() and pd_dict["type"] not in builtin_types.keys():
+        logger.warning("Type of pydata %s provided differs from specified type: %s", type(pydata).__name__, pd_dict["type"])
+        pd_dict["type"] = type(pydata).__name__
     if pd_dict["type"].lower() == "json":
         try:
             pydata = json.loads(pydata)
@@ -56,28 +72,35 @@ def parse_pydata(pd_dict: dict) -> bytes:
         pydata = eval(pydata)
         # except:
         #     pydata = pydata.encode()
-    elif pd_dict["type"].lower() == "int":
+    elif pd_dict["type"].lower() == "int" or isinstance(pydata, int):
         try:
             pydata = int(pydata)
+            pd_dict["type"] = "int"
         except:
             pydata = pydata.encode()
-    elif pd_dict["type"].lower() == "float":
+    elif pd_dict["type"].lower() == "float" or isinstance(pydata, float):
         try:
             pydata = float(pydata)
+            pd_dict["type"] = "float"
         except:
             pydata = pydata.encode()
-    elif pd_dict["type"].lower() == "boolean":
+    elif pd_dict["type"].lower() == "boolean" or isinstance(pydata, bool):
         try:
             pydata = bool(pydata)
+            pd_dict["type"] = "bool"
         except:
             pydata = pydata.encode()
     elif pd_dict["type"].lower() == "object":
         pydata = base64.b64decode(pydata.encode())
         try:
-            pydata = pickle.loads(pydata)
+            pydata = dill.loads(pydata)
         except:
             raise
-    return pickle.dumps(pydata)
+    elif pd_dict["type"].lower() == "raw":
+        pydata = dill.loads(base64.b64decode(pydata))
+        logger.debug("Returning pydata of type: %s", type(pydata))
+        # return pydata
+    return dill.dumps(pydata)
 
 
 ##
@@ -117,34 +140,30 @@ def initialize(self, **kwargs):
         """
         args = []
         pydata = None
+        # pdict = {}
         pdict = {"type": "raw"}  # initialize this value to enforce BytesIO
         self.data_type = pdict["type"]
         field_names = (
             [f["name"] for f in kwargs["fields"]] if "fields" in kwargs else []
         )
         if "pydata" in kwargs and not (
             "fields" in kwargs and "pydata" in field_names
-        ):  # means that is was passed directly
+        ):  # means that is was passed directly (e.g. from tests)
             pydata = kwargs.pop("pydata")
-            logger.debug("pydata value provided: %s, %s", pydata, kwargs)
-            try:  # test whether given value is valid
-                _ = pickle.loads(base64.b64decode(pydata))
-                pydata = base64.b64decode(pydata)
-            except:
-                pydata = None
+            pdict["value"] = pydata
+            pydata = parse_pydata(pdict)
         elif "fields" in kwargs and "pydata" in field_names:
             data_pos = field_names.index("pydata")
             pdict = kwargs["fields"][data_pos]
             pydata = parse_pydata(pdict)
-        if  pdict["type"].lower() in ["str","string"]:
-            self.data_type =  "String"
-            self._buf = io.StringIO(*args)
+        if pdict and pdict["type"].lower() in ["str","string"]:
+            self.data_type =  "String" if pydata else "raw"
         else:
-            self.data_type = pdict["type"]
-            self._buf = io.BytesIO(*args)
+            self.data_type = pdict["type"] if pdict else ""
         if pydata:
             args.append(pydata)
-            logger.debug("Loaded into memory: %s, %s", pydata, self.data_type)
+            logger.debug("Loaded into memory: %s, %s, %s", pydata, self.data_type, type(pydata))
+        self._buf = io.BytesIO(*args) if self.data_type != "String" else io.StringIO(*args)
         self.size = len(pydata) if pydata else 0
 
     def getIO(self):
@@ -230,7 +249,7 @@ def initialize(self, **kwargs):
             pydata = kwargs.pop("pydata")
             logger.debug("pydata value provided: %s", pydata)
             try:  # test whether given value is valid
-                _ = pickle.loads(base64.b64decode(pydata.encode("latin1")))
+                _ = dill.loads(base64.b64decode(pydata.encode("latin1")))
                 pydata = base64.b64decode(pydata.encode("latin1"))
             except:
                 pydata = None

daliuge-engine/dlg/data/io.py

@@ -20,6 +20,7 @@
 #    MA 02111-1307  USA
 #
 from abc import abstractmethod, ABCMeta
+import base64
 from http.client import HTTPConnection
 from multiprocessing.sharedctypes import Value
 from overrides import overrides
@@ -32,6 +33,7 @@
 from typing import Optional, Union
 
 from dlg import ngaslite
+from dlg.common import b2s
 
 if sys.version_info >= (3, 8):
     from dlg.shared_memory import DlgSharedMemory
@@ -251,16 +253,25 @@ def _open(self, **kwargs):
         elif self._mode == OpenMode.OPEN_READ:
             # TODO: potentially wasteful copy
             if isinstance(self._buf, io.StringIO):
+                self._desc = io.StringIO
                 return io.StringIO(self._buf.getvalue())
             return io.BytesIO(self._buf.getbuffer())
         else:
             raise ValueError()
 
     @overrides
     def _write(self, data, **kwargs) -> int:
-        if isinstance(data, str):
+        if isinstance(self._desc, io.BytesIO) and isinstance(data, str):
             data = bytes(data, encoding="utf8")
-        self._desc.write(data)
+        elif isinstance(self._desc, io.StringIO) and isinstance(data, bytes):
+            data = b2s(base64.b64encode(data))
+        elif isinstance(data, memoryview):
+            data = bytes(data)
+        try:
+            self._desc.write(data)
+        except Exception:
+            logger.debug("Writing of %s failed: %s", data)
+            raise
         return len(data)
 
     @overrides