From 926e5239f36ff653933b2a4ae8520e3a4a918cfe Mon Sep 17 00:00:00 2001 From: Ivan Khomyakov Date: Fri, 4 Mar 2022 09:32:49 -0800 Subject: [PATCH 01/11] add predict_fn_or_cls argument; add cluster parameters --- nucleus/deploy/client.py | 51 +++++++++++++++++++++++++++++++++------- 1 file changed, 42 insertions(+), 9 deletions(-) diff --git a/nucleus/deploy/client.py b/nucleus/deploy/client.py index d49c4326..18479582 100644 --- a/nucleus/deploy/client.py +++ b/nucleus/deploy/client.py @@ -82,8 +82,10 @@ def register_endpoint_auth_decorator(self, endpoint_auth_decorator_fn): def create_model_bundle( self, model_bundle_name: str, - load_predict_fn: Callable[[DeployModel_T], Callable[[Any], Any]], env_params: Dict[str, str], + *, + load_predict_fn: Optional[Callable[[DeployModel_T], Callable[[Any], Any]]] = None, + predict_fn_or_cls: Optional[Callable[[Any], Any]] = None, requirements: Optional[List[str]] = None, model: Optional[DeployModel_T] = None, load_model_fn: Optional[Callable[[], DeployModel_T]] = None, @@ -91,7 +93,7 @@ def create_model_bundle( ) -> ModelBundle: """ Grabs a s3 signed url and uploads a model bundle to Scale Deploy. - A model bundle consists of a "load_predict_fn" and exactly one of "model" or "load_model_fn", such that + A model bundle consists of a "predict_fn_or_cls" or "load_predict_fn" and exactly one of "model" or "load_model_fn", such that load_predict_fn(model) or load_predict_fn(load_model_fn()) @@ -103,6 +105,7 @@ def create_model_bundle( model: Typically a trained Neural Network, e.g. a Pytorch module load_model_fn: Function that when run, loads a model, e.g. a Pytorch module load_predict_fn: Function that when called with model, returns a function that carries out inference + predict_fn_or_cls: Function or a Callable class that runs inference on the call function. bundle_url: Only for self-hosted mode. Desired location of bundle. requirements: A list of python package requirements, e.g. ["tensorflow==2.3.0", "tensorflow-hub==0.11.0"]. If no list has been passed, will default to the currently @@ -117,11 +120,15 @@ def create_model_bundle( "tensorflow_version": Version of tensorflow, e.g. "2.3.0". Only applicable if framework_type is tensorflow """ - if (model is not None and load_model_fn is not None) or ( - model is None and load_model_fn is None - ): + check_args = [ + predict_fn_or_cls is not None, + model is not None, + load_model_fn is not None, + ] + + if (sum(check_args) != 1): raise ValueError( - "Exactly one of model and load_model_fn should be non-None" + "Exactly one of `model` or `load_model_fn` or `predict_fn_or_cls` should be non-None" ) # TODO should we try to catch when people intentionally pass both model and load_model_fn as None? @@ -139,7 +146,9 @@ def create_model_bundle( bundle_metadata = {} # Create bundle - if model is not None: + if predict_fn_or_cls: + bundle = predict_fn_or_cls + elif model is not None: bundle = dict(model=model, load_predict_fn=load_predict_fn) bundle_metadata["load_predict_fn"] = inspect.getsource( load_predict_fn @@ -198,6 +207,8 @@ def create_model_endpoint( min_workers: int, max_workers: int, per_worker: int, + aws_role: str, + results_s3_bucket: str, gpu_type: Optional[str] = None, overwrite_existing_endpoint: bool = False, endpoint_type: str = "async", @@ -214,6 +225,8 @@ def create_model_endpoint( gpus: Number of gpus each worker should get, e.g. 0, 1, etc. min_workers: Minimum number of workers for model endpoint max_workers: Maximum number of workers for model endpoint + aws_role: worker AWS role + results_s3_bucket: Result S3 URL per_worker: An autoscaling parameter. Use this to make a tradeoff between latency and costs, a lower per_worker will mean more workers are created for a given workload gpu_type: If specifying a non-zero number of gpus, this controls the type of gpu requested. Current options are @@ -237,6 +250,8 @@ def create_model_endpoint( max_workers=max_workers, per_worker=per_worker, endpoint_type=endpoint_type, + aws_role=aws_role, + results_s3_bucket=results_s3_bucket, ) if gpus == 0: del payload["gpu_type"] @@ -346,7 +361,12 @@ def sync_request( return resp["result_url"] def async_request( - self, endpoint_id: str, url: str, return_pickled: bool = True + self, + endpoint_id: str, + *, + url: Optional[str] = None, + kwargs: Dict[str, Any] = None, + return_pickled: bool = True, ) -> str: """ Not recommended to use this, instead we recommend to use functions provided by AsyncModelEndpoint. @@ -358,6 +378,9 @@ def async_request( endpoint_id: The id of the endpoint to make the request to url: A url that points to a file containing model input. Must be accessible by Scale Deploy, hence it needs to either be public or a signedURL. + Incompatible with the parameter `kwargs`. + kwargs: A dictionary that defines named arguments of the __call__ bundle function. + Incompatible with the parameter `url`. return_pickled: Whether the python object returned is pickled, or directly written to the file returned. Returns: @@ -365,8 +388,18 @@ def async_request( Example output: `abcabcab-cabc-abca-0123456789ab` """ + if url and kwargs: + raise ValueError( + "Exactly one of `url` or `kwargs` should be non-None" + ) + if url: + payload = dict(url=url) + else: + payload = dict(args=kwargs) + payload["return_pickled"] = return_pickled + resp = self.connection.post( - payload=dict(url=url, return_pickled=return_pickled), + payload=payload, route=f"{ASYNC_TASK_PATH}/{endpoint_id}", ) return resp["task_id"] From 69595ea2ba999bd15d3462804cd5409cd914fa07 Mon Sep 17 00:00:00 2001 From: Ivan Khomyakov Date: Fri, 4 Mar 2022 09:46:07 -0800 Subject: [PATCH 02/11] black --- nucleus/deploy/client.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/nucleus/deploy/client.py b/nucleus/deploy/client.py index 18479582..830fb451 100644 --- a/nucleus/deploy/client.py +++ b/nucleus/deploy/client.py @@ -84,7 +84,9 @@ def create_model_bundle( model_bundle_name: str, env_params: Dict[str, str], *, - load_predict_fn: Optional[Callable[[DeployModel_T], Callable[[Any], Any]]] = None, + load_predict_fn: Optional[ + Callable[[DeployModel_T], Callable[[Any], Any]] + ] = None, predict_fn_or_cls: Optional[Callable[[Any], Any]] = None, requirements: Optional[List[str]] = None, model: Optional[DeployModel_T] = None, @@ -126,7 +128,7 @@ def create_model_bundle( load_model_fn is not None, ] - if (sum(check_args) != 1): + if sum(check_args) != 1: raise ValueError( "Exactly one of `model` or `load_model_fn` or `predict_fn_or_cls` should be non-None" ) @@ -161,7 +163,8 @@ def create_model_bundle( load_predict_fn ) bundle_metadata["load_model_fn"] = inspect.getsource( - load_model_fn) # type: ignore + load_model_fn + ) # type: ignore serialized_bundle = cloudpickle.dumps(bundle) if self.is_self_hosted: From 54b4443f984014c939996c379348c83567d0ccb7 Mon Sep 17 00:00:00 2001 From: Ivan Khomyakov Date: Fri, 4 Mar 2022 15:15:11 -0800 Subject: [PATCH 03/11] add code bundle --- nucleus/deploy/client.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/nucleus/deploy/client.py b/nucleus/deploy/client.py index 830fb451..e23e0be5 100644 --- a/nucleus/deploy/client.py +++ b/nucleus/deploy/client.py @@ -150,6 +150,11 @@ def create_model_bundle( # Create bundle if predict_fn_or_cls: bundle = predict_fn_or_cls + if inspect.isfunction(predict_fn_or_cls): + source_code = inspect.getsource(predict_fn_or_cls) + else: + source_code = inspect.getsource(predict_fn_or_cls.__class__) + bundle_metadata["predict_fn_or_cls"] = source_code elif model is not None: bundle = dict(model=model, load_predict_fn=load_predict_fn) bundle_metadata["load_predict_fn"] = inspect.getsource( From aae49d35b4cb7c3ec9f210ef6695d2d10fcd48a9 Mon Sep 17 00:00:00 2001 From: Ivan Khomyakov Date: Mon, 7 Mar 2022 10:39:33 -0800 Subject: [PATCH 04/11] rm aws_role and results_s3_bucket as params --- nucleus/deploy/client.py | 6 ------ 1 file changed, 6 deletions(-) diff --git a/nucleus/deploy/client.py b/nucleus/deploy/client.py index ea9cbea4..d93cdfb4 100644 --- a/nucleus/deploy/client.py +++ b/nucleus/deploy/client.py @@ -241,8 +241,6 @@ def create_model_endpoint( min_workers: int, max_workers: int, per_worker: int, - aws_role: str, - results_s3_bucket: str, gpu_type: Optional[str] = None, overwrite_existing_endpoint: bool = False, endpoint_type: str = "async", @@ -259,8 +257,6 @@ def create_model_endpoint( gpus: Number of gpus each worker should get, e.g. 0, 1, etc. min_workers: Minimum number of workers for model endpoint max_workers: Maximum number of workers for model endpoint - aws_role: worker AWS role - results_s3_bucket: Result S3 URL per_worker: An autoscaling parameter. Use this to make a tradeoff between latency and costs, a lower per_worker will mean more workers are created for a given workload gpu_type: If specifying a non-zero number of gpus, this controls the type of gpu requested. Current options are @@ -283,8 +279,6 @@ def create_model_endpoint( max_workers=max_workers, per_worker=per_worker, endpoint_type=endpoint_type, - aws_role=aws_role, - results_s3_bucket=results_s3_bucket, ) if gpus == 0: del payload["gpu_type"] From d79f1336f5844b3edfaa474e3b71c8e4a9905647 Mon Sep 17 00:00:00 2001 From: Ivan Khomyakov Date: Mon, 7 Mar 2022 15:53:25 -0800 Subject: [PATCH 05/11] fix tests --- nucleus/deploy/client.py | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/nucleus/deploy/client.py b/nucleus/deploy/client.py index d93cdfb4..fb4d9ae9 100644 --- a/nucleus/deploy/client.py +++ b/nucleus/deploy/client.py @@ -171,28 +171,30 @@ def create_model_bundle( bundle_metadata = {} # Create bundle if predict_fn_or_cls: - bundle = predict_fn_or_cls + bundle_fn_or_cls = predict_fn_or_cls + serialized_bundle = cloudpickle.dumps(bundle_fn_or_cls) if inspect.isfunction(predict_fn_or_cls): source_code = inspect.getsource(predict_fn_or_cls) else: source_code = inspect.getsource(predict_fn_or_cls.__class__) bundle_metadata["predict_fn_or_cls"] = source_code elif model is not None: - bundle = dict(model=model, load_predict_fn=load_predict_fn) + bundle_func_1 = dict(model=model, load_predict_fn=load_predict_fn) + serialized_bundle = cloudpickle.dumps(bundle_func_1) bundle_metadata["load_predict_fn"] = inspect.getsource( - load_predict_fn + load_predict_fn # type: ignore ) else: - bundle = dict( + bundle_func_2 = dict( load_model_fn=load_model_fn, load_predict_fn=load_predict_fn ) + serialized_bundle = cloudpickle.dumps(bundle_func_2) bundle_metadata["load_predict_fn"] = inspect.getsource( - load_predict_fn + load_predict_fn # type: ignore ) bundle_metadata["load_model_fn"] = inspect.getsource( - load_model_fn - ) # type: ignore - serialized_bundle = cloudpickle.dumps(bundle) + load_model_fn # type: ignore + ) if self.is_self_hosted: if self.upload_bundle_fn is None: From 1a9176f28704f631e37de2f2ce4f2a0143aa846c Mon Sep 17 00:00:00 2001 From: Ivan Khomyakov Date: Mon, 7 Mar 2022 16:04:06 -0800 Subject: [PATCH 06/11] black --- nucleus/deploy/client.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/nucleus/deploy/client.py b/nucleus/deploy/client.py index fb4d9ae9..3bf1b356 100644 --- a/nucleus/deploy/client.py +++ b/nucleus/deploy/client.py @@ -182,7 +182,7 @@ def create_model_bundle( bundle_func_1 = dict(model=model, load_predict_fn=load_predict_fn) serialized_bundle = cloudpickle.dumps(bundle_func_1) bundle_metadata["load_predict_fn"] = inspect.getsource( - load_predict_fn # type: ignore + load_predict_fn # type: ignore ) else: bundle_func_2 = dict( @@ -190,10 +190,10 @@ def create_model_bundle( ) serialized_bundle = cloudpickle.dumps(bundle_func_2) bundle_metadata["load_predict_fn"] = inspect.getsource( - load_predict_fn # type: ignore + load_predict_fn # type: ignore ) bundle_metadata["load_model_fn"] = inspect.getsource( - load_model_fn # type: ignore + load_model_fn # type: ignore ) if self.is_self_hosted: From 28ac5fa09b28e3a9f75aab87a52d723d5890bcd6 Mon Sep 17 00:00:00 2001 From: Ivan Khomyakov Date: Fri, 11 Mar 2022 13:52:36 -0500 Subject: [PATCH 07/11] clarify docstring; check if model bundle is correct --- nucleus/deploy/client.py | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/nucleus/deploy/client.py b/nucleus/deploy/client.py index 3bf1b356..44fea04f 100644 --- a/nucleus/deploy/client.py +++ b/nucleus/deploy/client.py @@ -116,19 +116,19 @@ def create_model_bundle( ) -> ModelBundle: """ Grabs a s3 signed url and uploads a model bundle to Scale Deploy. - A model bundle consists of a "predict_fn_or_cls" or "load_predict_fn" and exactly one of "model" or "load_model_fn", such that - load_predict_fn(model) - or - load_predict_fn(load_model_fn()) - returns a function predict_fn that takes in model input and returns model output. - Pre/post-processing code can be included inside load_predict_fn/model. + + A model bundle consists of exactly {predict_fn_or_cls}, {load_predict_fn + model}, or {load_predict_fn + load_model_fn}. + Pre/post-processing code can be included inside load_predict_fn/model or in predict_fn_or_cls call. Parameters: model_bundle_name: Name of model bundle you want to create. This acts as a unique identifier. + predict_fn_or_cls: Function or a Callable class that runs end-to-end (pre/post processing and model inference) on the call. + I.e. `predict_fn_or_cls(REQUEST) -> RESPONSE`. model: Typically a trained Neural Network, e.g. a Pytorch module - load_model_fn: Function that when run, loads a model, e.g. a Pytorch module load_predict_fn: Function that when called with model, returns a function that carries out inference - predict_fn_or_cls: Function or a Callable class that runs inference on the call function. + I.e. `load_predict_fn(model) -> func; func(REQUEST) -> RESPONSE` + load_model_fn: Function that when run, loads a model, e.g. a Pytorch module + I.e. `load_predict_fn(load_model_fn()) -> func; func(REQUEST) -> RESPONSE` bundle_url: Only for self-hosted mode. Desired location of bundle. Overrides any value given by self.bundle_location_fn requirements: A list of python package requirements, e.g. @@ -146,13 +146,13 @@ def create_model_bundle( check_args = [ predict_fn_or_cls is not None, - model is not None, - load_model_fn is not None, + load_predict_fn is not None and model is not None, + load_predict_fn is not None and load_model_fn is not None, ] if sum(check_args) != 1: raise ValueError( - "Exactly one of `model` or `load_model_fn` or `predict_fn_or_cls` should be non-None" + "A model bundle consists of exactly {predict_fn_or_cls}, {load_predict_fn + model}, or {load_predict_fn + load_model_fn}." ) # TODO should we try to catch when people intentionally pass both model and load_model_fn as None? From 6641c1710b5759a666ca8859648ac16b2f16bcdb Mon Sep 17 00:00:00 2001 From: Ivan Khomyakov Date: Fri, 11 Mar 2022 19:26:19 -0500 Subject: [PATCH 08/11] use a single bundle var --- nucleus/deploy/client.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/nucleus/deploy/client.py b/nucleus/deploy/client.py index 44fea04f..57601e9b 100644 --- a/nucleus/deploy/client.py +++ b/nucleus/deploy/client.py @@ -168,27 +168,25 @@ def create_model_bundle( model_bundle_name, ) + bundle: Union[Callable[[Any], Any], Dict[str, Any], None] # validate bundle bundle_metadata = {} # Create bundle if predict_fn_or_cls: - bundle_fn_or_cls = predict_fn_or_cls - serialized_bundle = cloudpickle.dumps(bundle_fn_or_cls) + bundle = predict_fn_or_cls if inspect.isfunction(predict_fn_or_cls): source_code = inspect.getsource(predict_fn_or_cls) else: source_code = inspect.getsource(predict_fn_or_cls.__class__) bundle_metadata["predict_fn_or_cls"] = source_code elif model is not None: - bundle_func_1 = dict(model=model, load_predict_fn=load_predict_fn) - serialized_bundle = cloudpickle.dumps(bundle_func_1) + bundle = dict(model=model, load_predict_fn=load_predict_fn) bundle_metadata["load_predict_fn"] = inspect.getsource( load_predict_fn # type: ignore ) else: - bundle_func_2 = dict( + bundle = dict( load_model_fn=load_model_fn, load_predict_fn=load_predict_fn ) - serialized_bundle = cloudpickle.dumps(bundle_func_2) bundle_metadata["load_predict_fn"] = inspect.getsource( load_predict_fn # type: ignore ) @@ -196,6 +194,8 @@ def create_model_bundle( load_model_fn # type: ignore ) + serialized_bundle = cloudpickle.dumps(bundle) + if self.is_self_hosted: if self.upload_bundle_fn is None: raise ValueError("Upload_bundle_fn should be registered") From 5538b8561e6afdbb69104678f60653a286fa4aa8 Mon Sep 17 00:00:00 2001 From: Ivan Khomyakov Date: Fri, 11 Mar 2022 19:36:29 -0500 Subject: [PATCH 09/11] black --- nucleus/deploy/client.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/nucleus/deploy/client.py b/nucleus/deploy/client.py index 57601e9b..85a1b2bc 100644 --- a/nucleus/deploy/client.py +++ b/nucleus/deploy/client.py @@ -168,7 +168,9 @@ def create_model_bundle( model_bundle_name, ) - bundle: Union[Callable[[Any], Any], Dict[str, Any], None] # validate bundle + bundle: Union[ + Callable[[Any], Any], Dict[str, Any], None + ] # validate bundle bundle_metadata = {} # Create bundle if predict_fn_or_cls: From 51b6ff541b5a805bc2947ce353529f697bb14f49 Mon Sep 17 00:00:00 2001 From: Ivan Khomyakov Date: Sat, 12 Mar 2022 09:38:37 -0500 Subject: [PATCH 10/11] disable too-many-branches linter --- nucleus/deploy/client.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/nucleus/deploy/client.py b/nucleus/deploy/client.py index 0829fb51..c1c35c14 100644 --- a/nucleus/deploy/client.py +++ b/nucleus/deploy/client.py @@ -240,6 +240,8 @@ def create_model_bundle( "tensorflow_version": Version of tensorflow, e.g. "2.3.0". Only applicable if framework_type is tensorflow globals_copy: Dictionary of the global symbol table. Normally provided by `globals()` built-in function. """ + # TODO(ivan): remove `disable=too-many-branches` when get rid of `load_*` functions + #pylint: disable=too-many-branches check_args = [ predict_fn_or_cls is not None, From dc214070cd9627aa5783873dc0524603e6544942 Mon Sep 17 00:00:00 2001 From: Ivan Khomyakov Date: Sat, 12 Mar 2022 09:42:20 -0500 Subject: [PATCH 11/11] black --- nucleus/deploy/client.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/nucleus/deploy/client.py b/nucleus/deploy/client.py index c1c35c14..b83b03e1 100644 --- a/nucleus/deploy/client.py +++ b/nucleus/deploy/client.py @@ -241,7 +241,7 @@ def create_model_bundle( globals_copy: Dictionary of the global symbol table. Normally provided by `globals()` built-in function. """ # TODO(ivan): remove `disable=too-many-branches` when get rid of `load_*` functions - #pylint: disable=too-many-branches + # pylint: disable=too-many-branches check_args = [ predict_fn_or_cls is not None,