##### SUMMARY
An task using `async` and `poll` can fail intermittently due to an implementation flaw.

##### ISSUE TYPE
- Bug Report

##### COMPONENT NAME
lib/ansible/modules/utilities/logic/async_wrapper.py

##### ANSIBLE VERSION
```paste below
ansible --version
ansible 2.9.0.dev0
  config file = /etc/ansible/ansible.cfg
  configured module search path = [u'/root/.ansible/plugins/modules', u'/usr/share/ansible/plugins/modules']
  ansible python module location = /usr/lib/python2.7/site-packages/ansible
  executable location = /usr/bin/ansible
  python version = 2.7.5 (default, Jun 20 2019, 20:27:34) [GCC 4.8.5 20150623 (Red Hat 4.8.5-36)]
```

##### CONFIGURATION
```paste below
[root@awx awx_devel]# ansible-config dump --only-changed
[root@awx awx_devel]# 
```

##### OS / ENVIRONMENT
CentOS


##### STEPS TO REPRODUCE
Run this playbook:

https://github.com/ansible/test-playbooks/blob/master/async_tasks.yml

Against 5 hosts, where all host have hostvars

```yaml
ansible_host: 127.0.0.1
ansible_connection: local
```

This part of the playbook should be sufficient to reproduce error.

```yaml
---

- hosts: all
  gather_facts: false
  tasks:
  - name: Poll a sleep
    shell: "sleep 10"
    async: 30
    poll: 5
```


##### EXPECTED RESULTS
This should never error. There is no valid reason I can think of that this should ever error.


##### ACTUAL RESULTS
Extremely intermittent error on system that has good specs, but is running other servers and stuff while running the playbook. Output:

```paste below
Identity added: /tmp/awx_138_ji4bfrnh/artifacts/138/ssh_key_data (/tmp/awx_138_ji4bfrnh/artifacts/138/ssh_key_data)
SSH password: 
BECOME password[defaults to SSH password]: 

PLAY [all] *********************************************************************


TASK [Poll a sleep] ************************************************************





fatal: [HostGovernmentEffect94G]: FAILED! => {"ansible_facts": {"discovered_interpreter_python": "/usr/bin/python"}, "changed": true, "module_stderr": "", "module_stdout": "{\\"msg\\": \\"could not create: /tmp/.ansible_async\\", \\"failed\\": 1}\\n{\\"msg\\": \\"could not create: /tmp/.ansible_async\\", \\"failed\\": 1}\\n{\\"msg\\": \\"could not create: /tmp/.ansible_async\\", \\"failed\\": 1}\\n{\\"started\\": 1, \\"_ansible_suppress_tmpdir_delete\\": true, \\"finished\\": 0, \\"results_file\\": \\"/tmp/.ansible_async/478922372899.115\\", \\"ansible_job_id\\": \\"478922372899.115\\"}\\n", "msg": "MODULE FAILURE\\nSee stdout/stderr for the exact error", "rc": 0}
fatal: [HostBoneWorkingWX3]: FAILED! => {"ansible_facts": {"discovered_interpreter_python": "/usr/bin/python"}, "changed": true, "module_stderr": "", "module_stdout": "{\\"msg\\": \\"could not create: /tmp/.ansible_async\\", \\"failed\\": 1}\\n{\\"msg\\": \\"could not create: /tmp/.ansible_async\\", \\"failed\\": 1}\\n{\\"msg\\": \\"could not create: /tmp/.ansible_async\\", \\"failed\\": 1}\\n{\\"started\\": 1, \\"_ansible_suppress_tmpdir_delete\\": true, \\"finished\\": 0, \\"results_file\\": \\"/tmp/.ansible_async/263580704685.118\\", \\"ansible_job_id\\": \\"263580704685.118\\"}\\n", "msg": "MODULE FAILURE\\nSee stdout/stderr for the exact error", "rc": 0}
changed: [HostDreamObjectZEF]
changed: [HostShineInsideyaq]
changed: [HostQueenRecordUPH]

TASK [debug] *******************************************************************


ok: [HostQueenRecordUPH] => {
    "msg": "I'm a debug message."
}

ok: [HostDreamObjectZEF] => {
    "msg": "I'm a debug message."
}
ok: [HostShineInsideyaq] => {
    "msg": "I'm a debug message."
}

TASK [Fire and forget a slow command] ******************************************



changed: [HostShineInsideyaq]
changed: [HostDreamObjectZEF]
changed: [HostQueenRecordUPH]

TASK [debug] *******************************************************************



ok: [HostQueenRecordUPH] => {
    "msg": "I'm another debug message."
}
ok: [HostShineInsideyaq] => {
    "msg": "I'm another debug message."
}
ok: [HostDreamObjectZEF] => {
    "msg": "I'm another debug message."
}

TASK [Examine slow command] ****************************************************

FAILED - RETRYING: Examine slow command (20 retries left).


FAILED - RETRYING: Examine slow command (20 retries left).
FAILED - RETRYING: Examine slow command (20 retries left).
FAILED - RETRYING: Examine slow command (19 retries left).
FAILED - RETRYING: Examine slow command (19 retries left).
FAILED - RETRYING: Examine slow command (19 retries left).
FAILED - RETRYING: Examine slow command (18 retries left).
FAILED - RETRYING: Examine slow command (18 retries left).
FAILED - RETRYING: Examine slow command (18 retries left).
changed: [HostQueenRecordUPH]
changed: [HostShineInsideyaq]
changed: [HostDreamObjectZEF]

TASK [Fire and forget a slow reversal] *****************************************



changed: [HostDreamObjectZEF]
changed: [HostQueenRecordUPH]
changed: [HostShineInsideyaq]

TASK [debug] *******************************************************************
ok: [HostDreamObjectZEF] => {
    "msg": "I'm yet another debug message."
}



ok: [HostShineInsideyaq] => {
    "msg": "I'm yet another debug message."
}
ok: [HostQueenRecordUPH] => {
    "msg": "I'm yet another debug message."
}

TASK [Examine slow reversal] ***************************************************



FAILED - RETRYING: Examine slow reversal (20 retries left).
FAILED - RETRYING: Examine slow reversal (20 retries left).
FAILED - RETRYING: Examine slow reversal (20 retries left).
FAILED - RETRYING: Examine slow reversal (19 retries left).
FAILED - RETRYING: Examine slow reversal (19 retries left).
FAILED - RETRYING: Examine slow reversal (19 retries left).
changed: [HostQueenRecordUPH]
changed: [HostShineInsideyaq]
changed: [HostDreamObjectZEF]

PLAY RECAP *********************************************************************
HostBoneWorkingWX3         : ok=0    changed=0    unreachable=0    failed=1    skipped=0    rescued=0    ignored=0   
HostDreamObjectZEF         : ok=8    changed=5    unreachable=0    failed=0    skipped=0    rescued=0    ignored=0   
HostGovernmentEffect94G    : ok=0    changed=0    unreachable=0    failed=1    skipped=0    rescued=0    ignored=0   
HostQueenRecordUPH         : ok=8    changed=5    unreachable=0    failed=0    skipped=0    rescued=0    ignored=0   
HostShineInsideyaq         : ok=8    changed=5    unreachable=0    failed=0    skipped=0    rescued=0    ignored=0 
```

This hits the code:

https://github.com/ansible/ansible/blob/b2e992cecd93fbedc260d86fcb25bc39191e0b5b/lib/ansible/modules/utilities/logic/async_wrapper.py#L235-L242

The blanket catching of `Exception` with no reporting makes this extremely hard to diagnose. Debug options will not help the user here. I had to make a code change:

```diff
diff --git a/lib/ansible/modules/utilities/logic/async_wrapper.py b/lib/ansible/modules/utilities/logic/async_wrapper.py
index 586438ff13..183d857033 100644
--- a/lib/ansible/modules/utilities/logic/async_wrapper.py
+++ b/lib/ansible/modules/utilities/logic/async_wrapper.py
@@ -235,10 +235,10 @@ if __name__ == '__main__':
     if not os.path.exists(jobdir):
         try:
             os.makedirs(jobdir)
-        except Exception:
+        except Exception as e:
             print(json.dumps({
                 "failed": 1,
-                "msg": "could not create: %s" % jobdir
+                "msg": "could not create: %s: %s" % (jobdir, e)
             }))
     # immediately exit this process, leaving an orphaned process
     # running which immediately forks a supervisory timing process
```

With that, I was able to get this more helpful output:

```
TASK [Poll a sleep] ************************************************************





fatal: [HostQueenRecordUPH]: FAILED! => {"ansible_facts": {"discovered_interpreter_python": "/usr/bin/python"}, "changed": true, "module_stderr": "", "module_stdout": "{\\"msg\\": \\"could not create: /tmp/.ansible_async: [Errno 17] File exists: '/tmp/.ansible_async'\\", \\"failed\\": 1}\\n{\\"msg\\": \\"could not create: /tmp/.ansible_async: [Errno 17] File exists: '/tmp/.ansible_async'\\", \\"failed\\": 1}\\n{\\"msg\\": \\"could not create: /tmp/.ansible_async: [Errno 17] File exists: '/tmp/.ansible_async'\\", \\"failed\\": 1}\\n{\\"started\\": 1, \\"_ansible_suppress_tmpdir_delete\\": true, \\"finished\\": 0, \\"results_file\\": \\"/tmp/.ansible_async/265831391750.118\\", \\"ansible_job_id\\": \\"265831391750.118\\"}\\n", "msg": "MODULE FAILURE\\nSee stdout/stderr for the exact error", "rc": 0}
changed: [HostShineInsideyaq]
changed: [HostBoneWorkingWX3]
changed: [HostDreamObjectZEF]
changed: [HostGovernmentEffect94G]
```

I have also found, unsurprisingly, that running multiple versions of this playbook run simultaneously increases the likelihood and number of hosts experiencing this bug.

To sum up what this suggests:

 - The check for the async folder is non-atomic, which bodes poorly for running locally with >1 process
 - While the job folder name is properly unique, its parent folder (the async folder) is global, and the non-atomic check + creation will cause multiple processes to trip over each other
if len(sys.argv) < 5:
        print(json.dumps({
            "failed": True,
            "msg": "usage: async_wrapper <jid> <time_limit> <modulescript> <argsfile> [-preserve_tmp]  "
                   "Humans, do not call directly!"
        }))
        sys.exit(1)
    jid = "%s.%d" % (sys.argv[1], os.getpid())
    time_limit = sys.argv[2]
    wrapped_module = sys.argv[3]
    argsfile = sys.argv[4]
    if '-tmp-' not in os.path.dirname(wrapped_module):
        preserve_tmp = True
    elif len(sys.argv) > 5:
        preserve_tmp = sys.argv[5] == '-preserve_tmp'
    else:
        preserve_tmp = False
    
    if argsfile != '_':
        cmd = "%s %s" % (wrapped_module, argsfile)
    else:
        cmd = wrapped_module
    step = 5
    async_dir = os.environ.get('ANSIBLE_ASYNC_DIR', '~/.ansible_async')
    
    jobdir = os.path.expanduser(async_dir)
    job_path = os.path.join(jobdir, jid)
    if not os.path.exists(jobdir):