Support functools.partial with FunctionExecutor

Fixes #1428

Author: chuckwondo

CHANGELOG.md

@@ -6,16 +6,17 @@
 -
 
 ### Changed
-- 
+-
 
 ### Fixed
-- 
+- [Executor] Support use of `functools.partial` with FunctionExecutor's
+  `call_async` and `map` methods
 
 ## [v3.6.3]
 
 ### Fixed
-- Fixed memory available options for aws batch: 4 cpus 
-- Fixed race condition and improving monitor stability 
+- Fixed memory available options for aws batch: 4 cpus
+- Fixed race condition and improving monitor stability
 
 
 ## [v3.6.2]
@@ -103,7 +104,7 @@
 - [AWS Batch] Updated CPU and Memory resource limits
 
 ### Fixed
-- [AWS Lambda] Fixed wrong AWS Lambda delete runtime_name match semantics 
+- [AWS Lambda] Fixed wrong AWS Lambda delete runtime_name match semantics
 - [Worker] Fixed potential issue that can appear during 'func_obj' loading from cache
 - [Monitor] Fixed potential 'keyerror' exceptions
 - [Swift] Fixed OpenStack Swift parameters and authentication by adding domain information
@@ -175,7 +176,7 @@
 ### Added
 - [k8s] Added a new way of invoking functions using a RabbitMQ work queue
 - [IBM VPC] Added "zone" config parameter
-- [IBM Code Engine] Get and print an error message in case of container execution failure 
+- [IBM Code Engine] Get and print an error message in case of container execution failure
 
 ### Changed
 - [OpenWhisk] Updated default runtimes
@@ -723,7 +724,7 @@
 - [Config] Allow 'log_stream' and 'log_filename' keys in configuration
 - [Config] Allow 'runtime' being configured at serverless backend level
 - [Config] Allow 'invoke_pool_threads' being configured at serverless backend level
-- [Multiprocessing] Added generic Manager 
+- [Multiprocessing] Added generic Manager
 - [Kubernetes] Add kubernetes job backend
 - [CLI] Extended lithops cli with storage put, get, delete and list options
 - [Azure] Added missing azure functions backend methods
@@ -970,7 +971,7 @@
 - [Core] Generic compute client logic
 - [Core] IBM IAM service client lib
 - [Core] IBM VPC service client lib
-- [Docker] Docker backend compatible with IBM VPC VM 
+- [Docker] Docker backend compatible with IBM VPC VM
 
 ### Changed
 -  [Docker] Improved Docker executor
@@ -1599,8 +1600,8 @@
 - Docs updated
 - Storage separation
 - Project update. 'bx' and 'wsk' CLI tools are no longer necessary
-- Updated setup.py 
-- Deleted requirements.txt 
+- Updated setup.py
+- Deleted requirements.txt
 - Updated default_preinstalls
 
 ### Fixed
@@ -1658,7 +1659,7 @@ First release.
 - When a new *executor class* is instantiated, it is created a *storage_handler* used in the all PyWren execution.
 - Now it is possible to specify the **runtime** when the user instantiates the *executor class* instead of changing the config file every time (In the config file is specified the default runtime).
 - The **logging level** is now specified when the user instantiates the *executor class* instead of put it in the first line of the code within an env variable.
-- The PyWren code which is executed remotely as a wrapper of the function now uses the main storage handler as the rest of the PyWren code. In previous versions, PyWren creates a new storage client directly with *boto3* library instead of using pywren/storage/storage.py wrapper. 
+- The PyWren code which is executed remotely as a wrapper of the function now uses the main storage handler as the rest of the PyWren code. In previous versions, PyWren creates a new storage client directly with *boto3* library instead of using pywren/storage/storage.py wrapper.
 - Added support for multiple parameters in the functions which are executed remotely as a cloud functions. Previous versions just allows one parameter.
 - Eased the usage of the storage backend within a function. By simply specifying *storage_handler* as a parameter of the function, the user will get access to the storage backend.
 - Added a new method for retrieving the results of an execution called **fetch_all_resuslts()**. Previous PyWren versions already includes a method called *get_all_results()*, but this is a sequential method and it takes long time to retrieve all the results. It was also included a *wait()* class which is more similar to *get_all_results()* method, the main difference is that the new method is all based on *list the available objects in a bucket*, and it returns when all the tasks are finished. The new method also has the possibility to activate a progress bar in order to track the current status of the execution (really useful for larger executions).

lithops/job/serialize.py

@@ -24,7 +24,7 @@
 import cloudpickle
 from pathlib import Path
 from dis import Bytecode
-from functools import reduce
+from functools import partial, reduce
 from importlib import import_module
 from types import CodeType, FunctionType, ModuleType
 
@@ -151,6 +151,9 @@ def _module_inspect(self, obj):
                     for k, v in linspect.getmembers_static(param):
                         if inspect.isfunction(v) or (inspect.ismethod(v) and inspect.isfunction(v.__func__)):
                             worklist.append(v)
+        elif isinstance(obj, partial):
+            found_methods = ["__call__"]
+            worklist.append(obj.func)
         else:
             # The obj is the user's function but in form of a class
             found_methods = []
@@ -159,8 +162,10 @@ def _module_inspect(self, obj):
                     found_methods.append(k)
                     worklist.append(v)
             if "__call__" not in found_methods:
-                raise Exception('The class you passed as the function to '
-                                'run must contain the "__call__" method')
+                raise ValueError(
+                    "The class you passed as the function to "
+                    'run must contain the "__call__" method'
+                )
 
         # The worklist is only used for analyzing functions
         for fn in worklist:
@@ -199,7 +204,7 @@ def _module_inspect(self, obj):
                     elif inspect.iscode(v):
                         codeworklist.append(v)
 
-        return set([mod_name.split('.')[0] for mod_name in mods])
+        return {mod_name.split(".")[0] for mod_name in mods}
 
     def _inner_module_inspect(self, inst):
         """

lithops/tests/functions.py

@@ -38,6 +38,22 @@ def _func(x):
     return futures
 
 
+def lithops_return_futures_map_over_partial(x):
+    from functools import partial
+
+    def _func(x, y):
+        return x * y
+
+    fexec = lithops.FunctionExecutor()
+    futures = fexec.map(partial(_func, 2), range(x))
+
+    # this while loop is required to pass localhost tests on Windows
+    while not all(f.running or f.ready for f in futures):
+        time.sleep(0.1)
+
+    return futures
+
+
 def lithops_return_futures_call_async(x):
     def _func(x):
         return x + 1

lithops/tests/test_call_async.py

@@ -49,6 +49,14 @@ def test_set_iterdata(self):
         result = fexec.get_result()
         assert result == 10
 
+    def test_partial_function(self):
+        from functools import partial
+
+        fexec = lithops.FunctionExecutor(config=pytest.lithops_config)
+        fexec.call_async(partial(simple_map_function, 4), 6)
+        result = fexec.get_result()
+        assert result == 10
+
     def test_dict_iterdata(self):
         fexec = lithops.FunctionExecutor(config=pytest.lithops_config)
         fexec.call_async(simple_map_function, {'x': 2, 'y': 8})

lithops/tests/test_map.py

@@ -21,9 +21,10 @@
     hello_world,
     lithops_inside_lithops_map_function,
     lithops_return_futures_map,
+    lithops_return_futures_map_over_partial,
     lithops_return_futures_call_async,
     lithops_return_futures_map_multiple,
-    concat
+    concat,
 )
 
 
@@ -106,6 +107,12 @@ def test_lithops_return_futures_map(self):
         result = fexec.get_result()
         assert result == [1, 2, 3]
 
+    def test_lithops_return_futures_map_over_partial(self):
+        fexec = lithops.FunctionExecutor(config=pytest.lithops_config)
+        fexec.call_async(lithops_return_futures_map_over_partial, 3)
+        result = fexec.get_result()
+        assert result == [0, 2, 4]
+
     def test_lithops_return_futures_call_async(self):
         fexec = lithops.FunctionExecutor(config=pytest.lithops_config)
         fexec.call_async(lithops_return_futures_call_async, 3)