diff --git a/lib/ansible/executor/task_executor.py b/lib/ansible/executor/task_executor.py
index 0e1ebb600c..e6e4cc3148 100644
--- a/lib/ansible/executor/task_executor.py
+++ b/lib/ansible/executor/task_executor.py
@@ -67,6 +67,7 @@ class TaskExecutor:
         self._new_stdin         = new_stdin
         self._loader            = loader
         self._shared_loader_obj = shared_loader_obj
+        self._connection        = None
 
     def run(self):
         '''
@@ -361,8 +362,9 @@ class TaskExecutor:
                 self._task.args = variable_params
 
         # get the connection and the handler for this execution
-        self._connection = self._get_connection(variables=variables, templar=templar)
-        self._connection.set_host_overrides(host=self._host)
+        if not self._connection or not getattr(self._connection, '_connected', False):
+            self._connection = self._get_connection(variables=variables, templar=templar)
+            self._connection.set_host_overrides(host=self._host)
 
         self._handler = self._get_action_handler(connection=self._connection, templar=templar)
 
diff --git a/lib/ansible/plugins/connection/winrm.py b/lib/ansible/plugins/connection/winrm.py
index aa0c7b35e5..a506d39010 100644
--- a/lib/ansible/plugins/connection/winrm.py
+++ b/lib/ansible/plugins/connection/winrm.py
@@ -169,14 +169,16 @@ class Connection(ConnectionBase):
         rs = protocol.send_message(xmltodict.unparse(rq))
 
     def _winrm_exec(self, command, args=(), from_exec=False, stdin_iterator=None):
+        if not self.protocol:
+            self.protocol = self._winrm_connect()
+            self._connected = True
+        if not self.shell_id:
+            self.shell_id = self.protocol.open_shell(codepage=65001) # UTF-8
+            display.vvvvv('WINRM OPEN SHELL: %s' % self.shell_id, host=self._winrm_host)
         if from_exec:
             display.vvvvv("WINRM EXEC %r %r" % (command, args), host=self._winrm_host)
         else:
             display.vvvvvv("WINRM EXEC %r %r" % (command, args), host=self._winrm_host)
-        if not self.protocol:
-            self.protocol = self._winrm_connect()
-        if not self.shell_id:
-            self.shell_id = self.protocol.open_shell(codepage=65001) # UTF-8
         command_id = None
         try:
             stdin_push_failed = False
@@ -211,6 +213,7 @@ class Connection(ConnectionBase):
     def _connect(self):
         if not self.protocol:
             self.protocol = self._winrm_connect()
+            self._connected = True
         return self
 
     def exec_command(self, cmd, in_data=None, sudoable=True):
@@ -387,5 +390,8 @@ class Connection(ConnectionBase):
 
     def close(self):
         if self.protocol and self.shell_id:
+            display.vvvvv('WINRM CLOSE SHELL: %s' % self.shell_id, host=self._winrm_host)
             self.protocol.close_shell(self.shell_id)
-            self.shell_id = None
+        self.shell_id = None
+        self.protocol = None
+        self._connected = False
diff --git a/test/integration/roles/test_win_raw/tasks/main.yml b/test/integration/roles/test_win_raw/tasks/main.yml
index 6351c516be..30c1a75e6b 100644
--- a/test/integration/roles/test_win_raw/tasks/main.yml
+++ b/test/integration/roles/test_win_raw/tasks/main.yml
@@ -101,3 +101,16 @@
   assert:
     that:
       - "raw_result2.stdout_lines[0] == '--% icacls D:\\\\somedir\\\\ /grant \"! ЗАО. Руководство\":F'"
+
+# Assumes MaxShellsPerUser == 30 (the default)
+
+- name: test raw + with_items to verify that winrm connection is reused for each item
+  raw: echo "{{item}}"
+  with_items: "{{range(32)|list}}"
+  register: raw_with_items_result
+
+- name: check raw + with_items result
+  assert:
+    that:
+      - "not raw_with_items_result|failed"
+      - "raw_with_items_result.results|length == 32"