Remote function returning non-serializable type no longer shuts worker down (#384)

* Moved put_objects in main_loop to inside of try block

* Added test for failed serialization

* Fixed naming

* Minor

lib/python/ray/worker.py

@@ -964,6 +964,7 @@ def main_loop(worker=global_worker):
       outputs = worker.functions[function_name].executor(arguments) # execute the function
       if len(return_objectids) == 1:
         outputs = (outputs,)
+      store_outputs_in_objstore(return_objectids, outputs, worker) # store output in local object store
     except Exception as e:
       # If the task threw an exception, then record the traceback. We determine
       # whether the exception was thrown in the task execution by whether the
@@ -975,8 +976,6 @@ def main_loop(worker=global_worker):
       # Notify the scheduler that the task failed.
       raylib.notify_failure(worker.handle, function_name, str(failure_object), raylib.FailedTask)
       _logger().info("While running function {}, worker threw exception with message: \n\n{}\n".format(function_name, str(failure_object)))
-    else:
-      store_outputs_in_objstore(return_objectids, outputs, worker) # store output in local object store
     # Notify the scheduler that the task is done. This happens regardless of
     # whether the task succeeded or failed.
     raylib.ready_for_new_task(worker.handle)