Speed up task submissions a bit (#5992)

python/ray/_raylet.pyx

@@ -422,6 +422,9 @@ cdef deserialize_args(
     cdef:
         c_vector[shared_ptr[CRayObject]] by_reference_objects
 
+    if c_args.size() == 0:
+        return [], {}
+
     args = []
     by_reference_ids = []
     by_reference_indices = []

python/ray/memory_monitor.py

@@ -109,13 +109,13 @@ class MemoryMonitor(object):
         self.worker_name = worker_name
 
     def raise_if_low_memory(self):
-        if not psutil:
+        if psutil is None:
             return  # nothing we can do
 
-        if "RAY_DEBUG_DISABLE_MEMORY_MONITOR" in os.environ:
-            return  # escape hatch, not intended for user use
-
         if time.time() - self.last_checked > self.check_interval:
+            if "RAY_DEBUG_DISABLE_MEMORY_MONITOR" in os.environ:
+                return  # escape hatch, not intended for user use
+
             self.last_checked = time.time()
             total_gb = psutil.virtual_memory().total / (1024**3)
             used_gb = total_gb - psutil.virtual_memory().available / (1024**3)

python/ray/remote_function.py

@@ -57,6 +57,8 @@ class RemoteFunction(object):
                  object_store_memory, resources, num_return_vals, max_calls):
         self._function = function
         self._function_descriptor = FunctionDescriptor.from_function(function)
+        self._function_descriptor_list = (
+            self._function_descriptor.get_function_descriptor_list())
         self._function_name = (
             self._function.__module__ + "." + self._function.__name__)
         self._num_cpus = (DEFAULT_REMOTE_FUNCTION_CPUS
@@ -149,8 +151,8 @@ class RemoteFunction(object):
                     num_return_vals)
             else:
                 object_ids = worker.core_worker.submit_task(
-                    self._function_descriptor.get_function_descriptor_list(),
-                    list_args, num_return_vals, resources)
+                    self._function_descriptor_list, list_args, num_return_vals,
+                    resources)
 
             if len(object_ids) == 1:
                 return object_ids[0]

python/ray/utils.py

@@ -263,13 +263,22 @@ def get_cuda_visible_devices():
     return [int(i) for i in gpu_ids_str.split(",")]
 
 
+last_set_gpu_ids = None
+
+
 def set_cuda_visible_devices(gpu_ids):
     """Set the CUDA_VISIBLE_DEVICES environment variable.
 
     Args:
         gpu_ids: This is a list of integers representing GPU IDs.
     """
+
+    global last_set_gpu_ids
+    if last_set_gpu_ids == gpu_ids:
+        return  # optimization: already set
+
     os.environ["CUDA_VISIBLE_DEVICES"] = ",".join([str(i) for i in gpu_ids])
+    last_set_gpu_ids = gpu_ids
 
 
 def resources_from_resource_arguments(