### Summary

I want to do an action only once on a cluster, I execute a playbook with three hosts and serial :1.
The first host execute the action with meta: end_play at the end. The playbook continue for other hosts but I see meta: end play with verbose -vv iin runtime.
```yaml
- name: Backup postgresql data and push to aws with barman-cloud
  hosts:  "{{ target }}"
  become: true
  become_user: postgres
  serial: 1
  any_errors_fatal: true
  gather_facts: true
```

This is the end of the task I want to execute on one host
```yaml
- name: Remove file config file
  become_user: root
  file:
    path: "{{ item }}"
    state: absent
  with_items:
    - /tmp/config
    - /tmp/credentials
    - /etc/barman.conf
    - /etc/barman.d/

- name: Backup succeeded - End play
  meta: end_play
```
```
TASK [Remove file config file] ***************************************************************
changed: [int_postgres1] => (item=/tmp/config) => {"ansible_loop_var": "item", "changed": true, "item": "/tmp/config", "path": "/tmp/config", "state": "absent"}
changed: [int_postgres1] => (item=/tmp/credentials) => {"ansible_loop_var": "item", "changed": true, "item": "/tmp/credentials", "path": "/tmp/credentials", "state": "absent"}
changed: [int_postgres1] => (item=/etc/barman.conf) => {"ansible_loop_var": "item", "changed": true, "item": "/etc/barman.conf", "path": "/etc/barman.conf", "state": "absent"}
changed: [int_postgres1] => (item=/etc/barman.d/) => {"ansible_loop_var": "item", "changed": true, "item": "/etc/barman.d/", "path": "/etc/barman.d/", "state": "absent"}
META: ending play

PLAY [Backup postgresql data and push to aws with barman-cloud] ******************************
Friday 19 March 2021  20:20:03 +0100 (0:00:01.171)       0:00:18.459 ********** 

TASK [Gathering Facts] ***********************************************************************
ok: [int_postgres2]
META: ran handlers
Friday 19 March 2021  20:20:04 +0100 (0:00:01.490)       0:00:19.950 ********** 
Friday 19 March 2021  20:20:05 +0100 (0:00:00.059)       0:00:20.010 ********** 

TASK [Check if postgres server is replica] ***************************************************
changed: [int_postgres2] => {"changed": true, "cmd": "curl -s http://***:8008/patroni | jq -r .role", "delta": "0:00:00.017332", "end": "2021-03-19 20:20:05.290983", "rc": 0, "start": "2021-03-19 20:20:05.273651", "stderr": "", "stderr_lines": [], "stdout": "replica", "stdout_lines": ["replica"]}
META: end_host conditional evaluated to false, continuing execution for int_postgres2
Friday 19 March 2021  20:20:05 +0100 (0:00:00.344)       0:00:20.355 ********** 
```
EDIT: Also run_once doesnt work with serial: 1, all hosts will execute the command in the same playbook.
I removed serial: 1 and add run_once on all commands I wanted to execute only once but I keep this issue because I'm not sure this is the expected behaviour for meta and serial: 1.

Tell me if I am wrong.


### Issue Type

Bug Report

### Component Name

meta

### Ansible Version

```console (paste below)
$ ansible --version
ansible 2.10.3
```


### Configuration

```console (paste below)
$ ansible-config dump --only-changed

```


### OS / Environment

Ubuntu 18.04

### Steps to Reproduce

<!--- Paste example playbooks or commands between quotes below -->
```yaml (paste below)
  hosts:  "{{ target }}"
  become: true
  become_user: postgres
  serial: 1
  any_errors_fatal: true
  gather_facts: true
```


### Expected Results

end_play meta stop all hosts even if serial: 1 is specified

### Actual Results

```console (paste below)
playbook continu and doesnt stop with meta end_play
```

if hasattr(self, '_workers'):
            for attempts_remaining in range(C.WORKER_SHUTDOWN_POLL_COUNT - 1, -1, -1):
                if not any(worker_prc and worker_prc.is_alive() for worker_prc in self._workers):
                    break
                if attempts_remaining:
                    time.sleep(C.WORKER_SHUTDOWN_POLL_DELAY)
                else:
                    display.warning('One or more worker processes are still running and will be terminated.')
            for worker_prc in self._workers:
                if worker_prc and worker_prc.is_alive():
                    try:
                        worker_prc.terminate()
                    except AttributeError:
                        pass
    def clear_failed_hosts(self):
        self._failed_hosts = dict()
    def get_inventory(self):
        return self._inventory
    def get_variable_manager(self):
        return self._variable_manager
    def get_loader(self):