diff --git a/fabric/main.py b/fabric/main.py
index 0f91f20c..184ab017 100644
--- a/fabric/main.py
+++ b/fabric/main.py
@@ -157,31 +157,21 @@ def load_fabfile(path, importer=None):
     # Get directory and fabfile name
     directory, fabfile = os.path.split(path)
     # If the directory isn't in the PYTHONPATH, add it so our import will work
-    added_to_path = False
-    index = None
     if directory not in sys.path:
         sys.path.insert(0, directory)
-        added_to_path = True
     # If the directory IS in the PYTHONPATH, move it to the front temporarily,
     # otherwise other fabfiles -- like Fabric's own -- may scoop the intended
     # one.
     else:
         i = sys.path.index(directory)
         if i != 0:
-            # Store index for later restoration
-            index = i
             # Add to front, then remove from original position
             sys.path.insert(0, directory)
             del sys.path[i + 1]
     # Perform the import (trimming off the .py)
     imported = importer(os.path.splitext(fabfile)[0])
-    # Remove directory from path if we added it ourselves (just to be neat)
-    if added_to_path:
-        del sys.path[0]
-    # Put back in original index if we moved it
-    if index is not None:
-        sys.path.insert(index + 1, directory)
-        del sys.path[0]
+
+    # leave sys.path changed so that parallel mode can re-import tasks
 
     # Actually load tasks
     docstring, new_style, classic, default = load_tasks_from_module(imported)
diff --git a/fabric/tasks.py b/fabric/tasks.py
index 1b36f8ec..519e6316 100644
--- a/fabric/tasks.py
+++ b/fabric/tasks.py
@@ -2,6 +2,7 @@
 import six
 import sys
 import textwrap
+from importlib import import_module
 
 from fabric import state
 from fabric.utils import abort, warn, error
@@ -204,13 +205,15 @@ def _is_network_error_ignored():
     return not state.env.use_exceptions_for['network'] and state.env.skip_bad_hosts
 
 
-def _parallel_wrap(task, args, kwargs, queue, name, env):
+def _parallel_wrap(task_module, task_name, args, kwargs, queue, name, env):
     # Wrap in another callable that:
     # * expands the env it's given to ensure parallel, linewise, etc are
     #   all set correctly and explicitly
     # * nukes the connection cache to prevent shared-access problems
     # * knows how to send the tasks' return value back over a Queue
     # * captures exceptions raised by the task
+    module = import_module(task_module)
+    task = getattr(module, task_name)
     state.env.update(env)
     try:
         state.connections.clear()
@@ -250,7 +253,8 @@ def _execute(task, host, my_env, args, kwargs, jobs, queue, multiprocessing):
 
         # Stuff into Process wrapper
         kwarg_dict = {
-            'task': task,
+            'task_module': task.__module__,
+            'task_name': task.__name__,
             'args': args,
             'kwargs': kwargs,
             'queue': queue,