refactoring async

- centralized skipping
- also fixed module name broken by previous refactor
- let action modules handle async processing
- moved async into base action class's module exec
- action plugins can now run final action as async
- actually skip copy if base skips
- fixed normal for new paths
- ensure internal stat is never async
- default poll to 10 as per docs
- added hint for callback fix on poll
- restructured late tmp, now a pipeline query
- moving action handler to connection as networking does
- fixed network assumption invocation is always passed
- centralized key cleanup, normalized internal var
- _supress_tmpdir_delete now in _ansible_xxx and gets removed from results
- delay internal key removal till after we use em
- nicer tmp removing, using existing methods
- moved cleanup tmp flag to mking tmp func

lib/ansible/executor/task_executor.py

@@ -517,6 +517,7 @@ class TaskExecutor:
             if self._task.async > 0:
                 if self._task.poll > 0 and not result.get('skipped'):
                     result = self._poll_async_result(result=result, templar=templar, task_vars=vars_copy)
+                    #FIXME callback 'v2_runner_on_async_poll' here
 
                 # ensure no log is preserved
                 result["_ansible_no_log"] = self._play_context.no_log
@@ -757,15 +758,12 @@ class TaskExecutor:
         Returns the correct action plugin to handle the requestion task action
         '''
 
+        # let action plugin override module, fallback to 'normal' action plugin otherwise
         if self._task.action in self._shared_loader_obj.action_loader:
-            if self._task.async != 0:
-                raise AnsibleError("async mode is not supported with the %s module" % self._task.action)
             handler_name = self._task.action
-        elif self._task.async == 0:
+        else:
             pc_conn = self._shared_loader_obj.connection_loader.get(self._play_context.connection, class_only=True)
             handler_name = getattr(pc_conn, 'action_handler', 'normal')
-        else:
-            handler_name = 'async'
 
         handler = self._shared_loader_obj.action_loader.get(
             handler_name,