diff --git a/changelog/61153.added b/changelog/61153.added
new file mode 100644
index 000000000000..ad7b978a7f2f
--- /dev/null
+++ b/changelog/61153.added
@@ -0,0 +1 @@
+Initial work to allow parallel startup of proxy minions when used as sub proxies with Deltaproxy.
diff --git a/salt/metaproxy/deltaproxy.py b/salt/metaproxy/deltaproxy.py
index 9a914c84ae1c..206cfce5cf4f 100644
--- a/salt/metaproxy/deltaproxy.py
+++ b/salt/metaproxy/deltaproxy.py
@@ -2,7 +2,7 @@
 #   Proxy minion metaproxy modules
 #
 
-import copy
+import concurrent.futures
 import logging
 import os
 import signal
@@ -320,168 +320,224 @@ def post_master_init(self, master):
     self.proxy_pillar = {}
     self.proxy_context = {}
     self.add_periodic_callback("cleanup", self.cleanup_subprocesses)
-    for _id in self.opts["proxy"].get("ids", []):
-        control_id = self.opts["id"]
-        proxyopts = self.opts.copy()
-        proxyopts["id"] = _id
 
-        proxyopts = salt.config.proxy_config(
-            self.opts["conf_file"], defaults=proxyopts, minion_id=_id
-        )
-        proxyopts["id"] = proxyopts["proxyid"] = _id
+    if self.opts["proxy"].get("parallel_startup"):
+        log.debug("Initiating parallel startup for proxies")
+        with concurrent.futures.ThreadPoolExecutor() as executor:
+            futures = [
+                executor.submit(
+                    subproxy_post_master_init,
+                    _id,
+                    uid,
+                    self.opts,
+                    self.proxy,
+                    self.utils,
+                )
+                for _id in self.opts["proxy"].get("ids", [])
+            ]
+
+        for f in concurrent.futures.as_completed(futures):
+            sub_proxy_data = f.result()
+            minion_id = sub_proxy_data["proxy_opts"].get("id")
+
+            if sub_proxy_data["proxy_minion"]:
+                self.deltaproxy_opts[minion_id] = sub_proxy_data["proxy_opts"]
+                self.deltaproxy_objs[minion_id] = sub_proxy_data["proxy_minion"]
+
+                if self.deltaproxy_opts[minion_id] and self.deltaproxy_objs[minion_id]:
+                    self.deltaproxy_objs[
+                        minion_id
+                    ].req_channel = salt.transport.client.AsyncReqChannel.factory(
+                        sub_proxy_data["proxy_opts"], io_loop=self.io_loop
+                    )
+    else:
+        log.debug("Initiating non-parallel startup for proxies")
+        for _id in self.opts["proxy"].get("ids", []):
+            sub_proxy_data = subproxy_post_master_init(
+                _id, uid, self.opts, self.proxy, self.utils
+            )
 
-        proxyopts["subproxy"] = True
+            minion_id = sub_proxy_data["proxy_opts"].get("id")
 
-        self.proxy_context[_id] = {"proxy_id": _id}
+            if sub_proxy_data["proxy_minion"]:
+                self.deltaproxy_opts[minion_id] = sub_proxy_data["proxy_opts"]
+                self.deltaproxy_objs[minion_id] = sub_proxy_data["proxy_minion"]
 
-        # We need grains first to be able to load pillar, which is where we keep the proxy
-        # configurations
-        self.proxy_grains[_id] = salt.loader.grains(
-            proxyopts, proxy=self.proxy, context=self.proxy_context[_id]
-        )
-        self.proxy_pillar[_id] = yield salt.pillar.get_async_pillar(
-            proxyopts,
-            self.proxy_grains[_id],
-            _id,
-            saltenv=proxyopts["saltenv"],
-            pillarenv=proxyopts.get("pillarenv"),
-        ).compile_pillar()
+                if self.deltaproxy_opts[minion_id] and self.deltaproxy_objs[minion_id]:
+                    self.deltaproxy_objs[
+                        minion_id
+                    ].req_channel = salt.transport.client.AsyncReqChannel.factory(
+                        sub_proxy_data["proxy_opts"], io_loop=self.io_loop
+                    )
 
-        proxyopts["proxy"] = self.proxy_pillar[_id].get("proxy", {})
-        if not proxyopts["proxy"]:
-            log.warning(
-                "Pillar data for proxy minion %s could not be loaded, skipping.", _id
-            )
-            continue
+    self.ready = True
 
-        # Remove ids
-        proxyopts["proxy"].pop("ids", None)
 
-        proxyopts["pillar"] = self.proxy_pillar[_id]
-        proxyopts["grains"] = self.proxy_grains[_id]
+def subproxy_post_master_init(minion_id, uid, opts, main_proxy, main_utils):
+    """
+    Function to finish init after a deltaproxy proxy
+    minion has finished connecting to a master.
 
-        proxyopts["hash_id"] = self.opts["id"]
+    This is primarily loading modules, pillars, etc. (since they need
+    to know which master they connected to) for the sub proxy minions.
+    """
+    proxy_grains = {}
+    proxy_pillar = {}
 
-        _proxy_minion = ProxyMinion(proxyopts)
-        _proxy_minion.proc_dir = salt.minion.get_proc_dir(
-            proxyopts["cachedir"], uid=uid
-        )
+    proxyopts = opts.copy()
+    proxyopts["id"] = minion_id
 
-        _proxy_minion.proxy = salt.loader.proxy(
-            proxyopts, utils=self.utils, context=self.proxy_context[_id]
-        )
-        _proxy_minion.subprocess_list = self.subprocess_list
+    proxyopts = salt.config.proxy_config(
+        opts["conf_file"], defaults=proxyopts, minion_id=minion_id
+    )
+    proxyopts.update({"id": minion_id, "proxyid": minion_id, "subproxy": True})
 
-        # a long-running req channel
-        _proxy_minion.req_channel = salt.transport.client.AsyncReqChannel.factory(
-            proxyopts, io_loop=self.io_loop
-        )
+    proxy_context = {"proxy_id": minion_id}
 
-        # And load the modules
-        (
-            _proxy_minion.functions,
-            _proxy_minion.returners,
-            _proxy_minion.function_errors,
-            _proxy_minion.executors,
-        ) = _proxy_minion._load_modules(
-            opts=proxyopts, grains=proxyopts["grains"], context=self.proxy_context[_id]
+    # We need grains first to be able to load pillar, which is where we keep the proxy
+    # configurations
+    proxy_grains = salt.loader.grains(
+        proxyopts, proxy=main_proxy, context=proxy_context
+    )
+    proxy_pillar = salt.pillar.get_pillar(
+        proxyopts,
+        proxy_grains,
+        minion_id,
+        saltenv=proxyopts["saltenv"],
+        pillarenv=proxyopts.get("pillarenv"),
+    ).compile_pillar()
+
+    proxyopts["proxy"] = proxy_pillar.get("proxy", {})
+    if not proxyopts["proxy"]:
+        log.warning(
+            "Pillar data for proxy minion %s could not be loaded, skipping.", minion_id
         )
+        return {"proxy_minion": None, "proxy_opts": {}}
 
-        # we can then sync any proxymodules down from the master
-        # we do a sync_all here in case proxy code was installed by
-        # SPM or was manually placed in /srv/salt/_modules etc.
-        _proxy_minion.functions["saltutil.sync_all"](saltenv=self.opts["saltenv"])
+    # Remove ids
+    proxyopts["proxy"].pop("ids", None)
 
-        # And re-load the modules so the __proxy__ variable gets injected
-        (
-            _proxy_minion.functions,
-            _proxy_minion.returners,
-            _proxy_minion.function_errors,
-            _proxy_minion.executors,
-        ) = _proxy_minion._load_modules(
-            opts=proxyopts, grains=proxyopts["grains"], context=self.proxy_context[_id]
-        )
+    proxyopts.update(
+        {
+            "pillar": proxy_pillar,
+            "grains": proxy_grains,
+            "hash_id": opts["id"],
+        }
+    )
 
-        _proxy_minion.functions.pack["__proxy__"] = _proxy_minion.proxy
-        _proxy_minion.proxy.pack["__salt__"] = _proxy_minion.functions
-        _proxy_minion.proxy.pack["__ret__"] = _proxy_minion.returners
-        _proxy_minion.proxy.pack["__pillar__"] = proxyopts["pillar"]
-        _proxy_minion.proxy.pack["__grains__"] = proxyopts["grains"]
+    _proxy_minion = ProxyMinion(proxyopts)
+    _proxy_minion.proc_dir = salt.minion.get_proc_dir(proxyopts["cachedir"], uid=uid)
 
-        # Reload utils as well (chicken and egg, __utils__ needs __proxy__ and __proxy__ needs __utils__
-        _proxy_minion.proxy.utils = salt.loader.utils(
-            proxyopts, proxy=_proxy_minion.proxy, context=self.proxy_context[_id]
-        )
+    _proxy_minion.proxy = salt.loader.proxy(
+        proxyopts, utils=main_utils, context=proxy_context
+    )
+
+    # And load the modules
+    (
+        _proxy_minion.functions,
+        _proxy_minion.returners,
+        _proxy_minion.function_errors,
+        _proxy_minion.executors,
+    ) = _proxy_minion._load_modules(
+        opts=proxyopts,
+        grains=proxyopts["grains"],
+        context=proxy_context,
+    )
 
-        _proxy_minion.proxy.pack["__utils__"] = _proxy_minion.proxy.utils
+    # we can then sync any proxymodules down from the master
+    # we do a sync_all here in case proxy code was installed by
+    # SPM or was manually placed in /srv/salt/_modules etc.
+    _proxy_minion.functions["saltutil.sync_all"](saltenv=opts["saltenv"])
 
-        # Reload all modules so all dunder variables are injected
-        _proxy_minion.proxy.reload_modules()
+    # And re-load the modules so the __proxy__ variable gets injected
+    (
+        _proxy_minion.functions,
+        _proxy_minion.returners,
+        _proxy_minion.function_errors,
+        _proxy_minion.executors,
+    ) = _proxy_minion._load_modules(
+        opts=proxyopts,
+        grains=proxyopts["grains"],
+        context=proxy_context,
+    )
 
-        _proxy_minion.connected = True
+    _proxy_minion.functions.pack["__proxy__"] = _proxy_minion.proxy
+    _proxy_minion.proxy.pack["__salt__"] = _proxy_minion.functions
+    _proxy_minion.proxy.pack["__ret__"] = _proxy_minion.returners
+    _proxy_minion.proxy.pack["__pillar__"] = proxyopts["pillar"]
+    _proxy_minion.proxy.pack["__grains__"] = proxyopts["grains"]
 
-        _fq_proxyname = proxyopts["proxy"]["proxytype"]
+    # Reload utils as well (chicken and egg, __utils__ needs __proxy__ and __proxy__ needs __utils__
+    _proxy_minion.proxy.utils = salt.loader.utils(
+        proxyopts, proxy=_proxy_minion.proxy, context=proxy_context
+    )
 
-        proxy_init_fn = _proxy_minion.proxy[_fq_proxyname + ".init"]
-        try:
-            proxy_init_fn(proxyopts)
-        except Exception as exc:  # pylint: disable=broad-except
-            log.error(
-                "An exception occured during the initialization of minion %s: %s",
-                _id,
-                exc,
-                exc_info=True,
-            )
-            continue
+    _proxy_minion.proxy.pack["__utils__"] = _proxy_minion.proxy.utils
 
-        # Reload the grains
-        self.proxy_grains[_id] = salt.loader.grains(
-            proxyopts, proxy=_proxy_minion.proxy, context=self.proxy_context[_id]
+    # Reload all modules so all dunder variables are injected
+    _proxy_minion.proxy.reload_modules()
+
+    _proxy_minion.connected = True
+
+    _fq_proxyname = proxyopts["proxy"]["proxytype"]
+
+    proxy_init_fn = _proxy_minion.proxy[_fq_proxyname + ".init"]
+    try:
+        proxy_init_fn(proxyopts)
+    except Exception as exc:  # pylint: disable=broad-except
+        log.error(
+            "An exception occured during the initialization of minion %s: %s",
+            minion_id,
+            exc,
+            exc_info=True,
         )
-        proxyopts["grains"] = self.proxy_grains[_id]
-
-        if not hasattr(_proxy_minion, "schedule"):
-            _proxy_minion.schedule = salt.utils.schedule.Schedule(
-                proxyopts,
-                _proxy_minion.functions,
-                _proxy_minion.returners,
-                cleanup=[salt.minion.master_event(type="alive")],
-                proxy=_proxy_minion.proxy,
-                new_instance=True,
-                _subprocess_list=_proxy_minion.subprocess_list,
-            )
+        return {"proxy_minion": None, "proxy_opts": {}}
 
-        self.deltaproxy_objs[_id] = _proxy_minion
-        self.deltaproxy_opts[_id] = copy.deepcopy(proxyopts)
+    # Reload the grains
+    proxy_grains = salt.loader.grains(
+        proxyopts, proxy=_proxy_minion.proxy, context=proxy_context
+    )
+    proxyopts["grains"] = proxy_grains
 
-        # proxy keepalive
-        _proxy_alive_fn = _fq_proxyname + ".alive"
-        if (
-            _proxy_alive_fn in _proxy_minion.proxy
-            and "status.proxy_reconnect" in self.deltaproxy_objs[_id].functions
-            and proxyopts.get("proxy_keep_alive", True)
-        ):
-            # if `proxy_keep_alive` is either not specified, either set to False does not retry reconnecting
-            _proxy_minion.schedule.add_job(
-                {
-                    "__proxy_keepalive": {
-                        "function": "status.proxy_reconnect",
-                        "minutes": proxyopts.get(
-                            "proxy_keep_alive_interval", 1
-                        ),  # by default, check once per minute
-                        "jid_include": True,
-                        "maxrunning": 1,
-                        "return_job": False,
-                        "kwargs": {"proxy_name": _fq_proxyname},
-                    }
-                },
-                persist=True,
-            )
-            _proxy_minion.schedule.enable_schedule()
-        else:
-            _proxy_minion.schedule.delete_job("__proxy_keepalive", persist=True)
+    if not hasattr(_proxy_minion, "schedule"):
+        _proxy_minion.schedule = salt.utils.schedule.Schedule(
+            proxyopts,
+            _proxy_minion.functions,
+            _proxy_minion.returners,
+            cleanup=[salt.minion.master_event(type="alive")],
+            proxy=_proxy_minion.proxy,
+            new_instance=True,
+            _subprocess_list=_proxy_minion.subprocess_list,
+        )
 
-    self.ready = True
+    # proxy keepalive
+    _proxy_alive_fn = _fq_proxyname + ".alive"
+    if (
+        _proxy_alive_fn in _proxy_minion.proxy
+        and "status.proxy_reconnect" in _proxy_minion.functions
+        and proxyopts.get("proxy_keep_alive", True)
+    ):
+        # if `proxy_keep_alive` is either not specified, either set to False does not retry reconnecting
+        _proxy_minion.schedule.add_job(
+            {
+                "__proxy_keepalive": {
+                    "function": "status.proxy_reconnect",
+                    "minutes": proxyopts.get(
+                        "proxy_keep_alive_interval", 1
+                    ),  # by default, check once per minute
+                    "jid_include": True,
+                    "maxrunning": 1,
+                    "return_job": False,
+                    "kwargs": {"proxy_name": _fq_proxyname},
+                }
+            },
+            persist=True,
+        )
+        _proxy_minion.schedule.enable_schedule()
+    else:
+        _proxy_minion.schedule.delete_job("__proxy_keepalive", persist=True)
+
+    return {"proxy_minion": _proxy_minion, "proxy_opts": proxyopts}
 
 
 def target(cls, minion_instance, opts, data, connected):
@@ -1031,9 +1087,30 @@ def tune_in(self, start=True):
     Lock onto the publisher. This is the main event loop for the minion
     :rtype : None
     """
-    for proxy_id in self.deltaproxy_objs:
-        _proxy_minion = self.deltaproxy_objs[proxy_id]
-        _proxy_minion.setup_scheduler()
-        _proxy_minion.setup_beacons()
-        _proxy_minion._state_run()
+    if self.opts["proxy"].get("parallel_startup"):
+        with concurrent.futures.ThreadPoolExecutor() as executor:
+            futures = [
+                executor.submit(subproxy_tune_in, self.deltaproxy_objs[proxy_minion])
+                for proxy_minion in self.deltaproxy_objs
+            ]
+
+        for f in concurrent.futures.as_completed(futures):
+            _proxy_minion = f.result()
+            log.debug("Tune in for sub proxy %r finished", _proxy_minion.opts.get("id"))
+    else:
+        for proxy_minion in self.deltaproxy_objs:
+            _proxy_minion = subproxy_tune_in(self.deltaproxy_objs[proxy_minion])
+            log.debug("Tune in for sub proxy %r finished", _proxy_minion.opts.get("id"))
     super(ProxyMinion, self).tune_in(start=start)
+
+
+def subproxy_tune_in(proxy_minion, start=True):
+    """
+    Tunein sub proxy minions
+    """
+    proxy_minion.setup_scheduler()
+    proxy_minion.setup_beacons()
+    proxy_minion.add_periodic_callback("cleanup", proxy_minion.cleanup_subprocesses)
+    proxy_minion._state_run()
+
+    return proxy_minion
diff --git a/salt/minion.py b/salt/minion.py
index 544805fc6210..7afda943162a 100644
--- a/salt/minion.py
+++ b/salt/minion.py
@@ -3799,6 +3799,16 @@ def _post_master_init(self, master):
         mp_call = _metaproxy_call(self.opts, "post_master_init")
         return mp_call(self, master)
 
+    @salt.ext.tornado.gen.coroutine
+    def subproxy_post_master_init(self, minion_id, uid):
+        """
+        Function to finish init for the sub proxies
+
+        :rtype : None
+        """
+        mp_call = _metaproxy_call(self.opts, "subproxy_post_master_init")
+        return mp_call(self, minion_id, uid)
+
     def tune_in(self, start=True):
         """
         Lock onto the publisher. This is the main event loop for the minion
diff --git a/salt/proxy/ssh_sample.py b/salt/proxy/ssh_sample.py
index 6e7e31aa8787..eb4c975d6acd 100644
--- a/salt/proxy/ssh_sample.py
+++ b/salt/proxy/ssh_sample.py
@@ -8,14 +8,12 @@
 import logging
 
 import salt.utils.json
+import salt.utils.vt_helper
 from salt.utils.vt import TerminalException
-from salt.utils.vt_helper import SSHConnection
 
 # This must be present or the Salt loader won't load this module
 __proxyenabled__ = ["ssh_sample"]
 
-DETAILS = {}
-
 log = logging.getLogger(__file__)
 
 
@@ -36,13 +34,13 @@ def init(opts):
     Can be used to initialize the server connection.
     """
     try:
-        DETAILS["server"] = SSHConnection(
+        __context__["server"] = salt.utils.vt_helper.SSHConnection(
             host=__opts__["proxy"]["host"],
             username=__opts__["proxy"]["username"],
             password=__opts__["proxy"]["password"],
         )
-        out, err = DETAILS["server"].sendline("help")
-        DETAILS["initialized"] = True
+        out, err = __context__["server"].sendline("help")
+        __context__["initialized"] = True
 
     except TerminalException as e:
         log.error(e)
@@ -55,7 +53,7 @@ def initialized():
     places occur before the proxy can be initialized, return whether
     our init() function has been called
     """
-    return DETAILS.get("initialized", False)
+    return __context__.get("initialized", False)
 
 
 def grains():
@@ -63,23 +61,23 @@ def grains():
     Get the grains from the proxied device
     """
 
-    if not DETAILS.get("grains_cache", {}):
+    if not __context__.get("grains_cache", {}):
         cmd = "info"
 
         # Send the command to execute
-        out, err = DETAILS["server"].sendline(cmd)
+        out, err = __context__["server"].sendline(cmd)
 
         # "scrape" the output and return the right fields as a dict
-        DETAILS["grains_cache"] = parse(out)
+        __context__["grains_cache"] = parse(out)
 
-    return DETAILS["grains_cache"]
+    return __context__["grains_cache"]
 
 
 def grains_refresh():
     """
     Refresh the grains from the proxied device
     """
-    DETAILS["grains_cache"] = None
+    __context__["grains_cache"] = None
     return grains()
 
 
@@ -101,7 +99,7 @@ def ping():
     Ping the device on the other end of the connection
     """
     try:
-        out, err = DETAILS["server"].sendline("help")
+        out, err = __context__["server"].sendline("help")
         return True
     except TerminalException as e:
         log.error(e)
@@ -112,7 +110,7 @@ def shutdown(opts):
     """
     Disconnect
     """
-    DETAILS["server"].close_connection()
+    __context__["server"].close_connection()
 
 
 def parse(out):
@@ -146,7 +144,7 @@ def package_list():
 
     """
     # Send the command to execute
-    out, err = DETAILS["server"].sendline("pkg_list\n")
+    out, err = __context__["server"].sendline("pkg_list\n")
 
     # "scrape" the output and return the right fields as a dict
     return parse(out)
@@ -161,7 +159,7 @@ def package_install(name, **kwargs):
         cmd += " " + kwargs["version"]
 
     # Send the command to execute
-    out, err = DETAILS["server"].sendline(cmd)
+    out, err = __context__["server"].sendline(cmd)
 
     # "scrape" the output and return the right fields as a dict
     return parse(out)
@@ -174,7 +172,7 @@ def package_remove(name):
     cmd = "pkg_remove " + name
 
     # Send the command to execute
-    out, err = DETAILS["server"].sendline(cmd)
+    out, err = __context__["server"].sendline(cmd)
 
     # "scrape" the output and return the right fields as a dict
     return parse(out)
@@ -189,7 +187,7 @@ def service_list():
     cmd = "ps"
 
     # Send the command to execute
-    out, err = DETAILS["server"].sendline(cmd)
+    out, err = __context__["server"].sendline(cmd)
 
     # "scrape" the output and return the right fields as a dict
     return parse(out)
@@ -204,7 +202,7 @@ def service_start(name):
     cmd = "start " + name
 
     # Send the command to execute
-    out, err = DETAILS["server"].sendline(cmd)
+    out, err = __context__["server"].sendline(cmd)
 
     # "scrape" the output and return the right fields as a dict
     return parse(out)
@@ -219,7 +217,7 @@ def service_stop(name):
     cmd = "stop " + name
 
     # Send the command to execute
-    out, err = DETAILS["server"].sendline(cmd)
+    out, err = __context__["server"].sendline(cmd)
 
     # "scrape" the output and return the right fields as a dict
     return parse(out)
@@ -234,7 +232,7 @@ def service_restart(name):
     cmd = "restart " + name
 
     # Send the command to execute
-    out, err = DETAILS["server"].sendline(cmd)
+    out, err = __context__["server"].sendline(cmd)
 
     # "scrape" the output and return the right fields as a dict
     return parse(out)
diff --git a/tests/pytests/integration/cli/test_salt_deltaproxy.py b/tests/pytests/integration/cli/test_salt_deltaproxy.py
index 61823fb6bcd9..59bf3c59fae9 100644
--- a/tests/pytests/integration/cli/test_salt_deltaproxy.py
+++ b/tests/pytests/integration/cli/test_salt_deltaproxy.py
@@ -116,10 +116,16 @@ def test_exit_status_unknown_argument(salt_master, proxy_minion_id):
 # Hangs on Windows. You can add a timeout to the proxy.run command, but then
 # it just times out.
 @pytest.mark.skip_on_windows(reason=PRE_PYTEST_SKIP_REASON)
+@pytest.mark.parametrize(
+    "parallel_startup",
+    [True, False],
+    ids=["parallel_startup=True", "parallel_startup=False"],
+)
 def test_exit_status_correct_usage(
     salt_master,
     salt_cli,
     proxy_minion_id,
+    parallel_startup,
 ):
     """
     Ensure the salt-proxy control proxy starts and
@@ -153,11 +159,12 @@ def test_exit_status_correct_usage(
     controlproxy_pillar_file = """
     proxy:
         proxytype: deltaproxy
+        parallel_startup: {}
         ids:
           - {}
           - {}
     """.format(
-        proxy_one, proxy_two
+        parallel_startup, proxy_one, proxy_two
     )
 
     dummy_proxy_one_pillar_file = """
@@ -227,10 +234,16 @@ def test_exit_status_correct_usage(
 # Hangs on Windows. You can add a timeout to the proxy.run command, but then
 # it just times out.
 @pytest.mark.skip_on_windows(reason=PRE_PYTEST_SKIP_REASON)
+@pytest.mark.parametrize(
+    "parallel_startup",
+    [True, False],
+    ids=["parallel_startup=True", "parallel_startup=False"],
+)
 def test_missing_pillar_file(
     salt_master,
     salt_cli,
     proxy_minion_id,
+    parallel_startup,
 ):
     """
     Ensure that the control proxy minion starts up when
@@ -258,11 +271,12 @@ def test_missing_pillar_file(
     controlproxy_pillar_file = """
     proxy:
         proxytype: deltaproxy
+        parallel_startup: {}
         ids:
           - {}
           - {}
     """.format(
-        proxy_one, proxy_two
+        parallel_startup, proxy_one, proxy_two
     )
 
     dummy_proxy_one_pillar_file = """
@@ -318,10 +332,16 @@ def test_missing_pillar_file(
 # Hangs on Windows. You can add a timeout to the proxy.run command, but then
 # it just times out.
 @pytest.mark.skip_on_windows(reason=PRE_PYTEST_SKIP_REASON)
+@pytest.mark.parametrize(
+    "parallel_startup",
+    [True, False],
+    ids=["parallel_startup=True", "parallel_startup=False"],
+)
 def test_invalid_connection(
     salt_master,
     salt_cli,
     proxy_minion_id,
+    parallel_startup,
 ):
     """
     Ensure that the control proxy minion starts up when
@@ -356,12 +376,13 @@ def test_invalid_connection(
     controlproxy_pillar_file = """
     proxy:
         proxytype: deltaproxy
+        parallel_startup: {}
         ids:
           - {}
           - {}
           - {}
     """.format(
-        broken_proxy_one, broken_proxy_two, proxy_one
+        parallel_startup, broken_proxy_one, broken_proxy_two, proxy_one
     )
 
     dummy_proxy_one_pillar_file = """
diff --git a/tests/pytests/integration/proxy/conftest.py b/tests/pytests/integration/proxy/conftest.py
index 7406e59f5995..d924f4eba8ad 100644
--- a/tests/pytests/integration/proxy/conftest.py
+++ b/tests/pytests/integration/proxy/conftest.py
@@ -12,7 +12,16 @@ def salt_proxy(salt_master, salt_proxy_factory):
 
 
 @pytest.fixture(scope="module")
-def deltaproxy_pillar_tree(salt_master, salt_delta_proxy_factory):
+def deltaproxy_parallel_startup():
+    yield from [True, False]
+
+
+@pytest.fixture(
+    scope="module",
+    params=[True, False],
+    ids=["parallel_startup=True", "parallel_startup=False"],
+)
+def deltaproxy_pillar_tree(request, salt_master, salt_delta_proxy_factory):
     """
     Create the pillar files for controlproxy and two dummy proxy minions
     """
@@ -45,12 +54,14 @@ def deltaproxy_pillar_tree(salt_master, salt_delta_proxy_factory):
     controlproxy_pillar_file = """
     proxy:
         proxytype: deltaproxy
+        parallel_startup: {}
         ids:
           - {}
           - {}
           - {}
           - {}
     """.format(
+        request.param,
         proxy_one,
         proxy_two,
         proxy_three,
diff --git a/tests/pytests/integration/proxy/test_deltaproxy.py b/tests/pytests/integration/proxy/test_deltaproxy.py
index 01231334c990..4ea69e0b8a51 100644
--- a/tests/pytests/integration/proxy/test_deltaproxy.py
+++ b/tests/pytests/integration/proxy/test_deltaproxy.py
@@ -58,6 +58,16 @@ def test_list_pkgs(salt_cli, proxy_id):
     assert "redbull" in ret.data
 
 
+def test_list_pkgs_all(salt_cli, proxy_ids):
+    """
+    Ensure the proxy can ping (all proxy minions)
+    """
+    pkg_list = {"apache": "2.4", "coreutils": "1.0", "redbull": "999.99", "tinc": "1.4"}
+    ret = salt_cli.run("-L", "pkg.list_pkgs", minion_tgt=",".join(proxy_ids))
+    for _id in proxy_ids:
+        assert ret.data[_id] == pkg_list
+
+
 def test_install_pkgs(salt_cli, proxy_id):
     """
     Package test 2, really just tests that the virtual function capability
@@ -74,10 +84,44 @@ def test_install_pkgs(salt_cli, proxy_id):
     assert ret.data["thispkg"] == "1.0"
 
 
+def test_install_pkgs_all(salt_cli, proxy_ids):
+    """
+    Ensure the proxy can ping (all proxy minions)
+    """
+    install_ret = salt_cli.run(
+        "-L", "pkg.install", "thispkg", minion_tgt=",".join(proxy_ids)
+    )
+    list_ret = salt_cli.run("-L", "pkg.list_pkgs", minion_tgt=",".join(proxy_ids))
+
+    for _id in proxy_ids:
+
+        assert install_ret.data[_id]["thispkg"] == "1.0"
+
+        assert list_ret.data[_id]["apache"] == "2.4"
+        assert list_ret.data[_id]["redbull"] == "999.99"
+        assert list_ret.data[_id]["thispkg"] == "1.0"
+
+
 def test_remove_pkgs(salt_cli, proxy_id):
     ret = salt_cli.run("pkg.remove", "apache", minion_tgt=proxy_id)
     assert "apache" not in ret.data
 
+    # reinstall
+    ret = salt_cli.run("pkg.install", "apache", minion_tgt=proxy_id)
+
+
+def test_remove_pkgs_all(salt_cli, proxy_ids):
+    """
+    Ensure the proxy can ping (all proxy minions)
+    """
+    ret = salt_cli.run("-L", "pkg.remove", "coreutils", minion_tgt=",".join(proxy_ids))
+
+    for _id in proxy_ids:
+        assert "coreutils" not in ret.data[_id]
+
+    # reinstall
+    salt_cli.run("-L", "pkg.install", "coreutils", minion_tgt=",".join(proxy_ids))
+
 
 def test_upgrade(salt_cli, proxy_id):
     ret = salt_cli.run("pkg.upgrade", minion_tgt=proxy_id)
@@ -166,3 +210,64 @@ def test_config_get(salt_cli, proxy_id):
     """
     ret = salt_cli.run("config.get", "id", minion_tgt=proxy_id)
     assert ret.data == proxy_id
+
+
+def test_schedule_list(salt_cli, proxy_id):
+    """
+    Ensure schedule.list works
+    """
+    ret = salt_cli.run("schedule.list", minion_tgt=proxy_id)
+    assert ret.data == "schedule: {}\n"
+
+
+def test_schedule_add_list(salt_cli, proxy_id):
+    """
+    Ensure schedule.add works
+    """
+    ret = salt_cli.run(
+        "schedule.add", name="job1", function="test.ping", minion_tgt=proxy_id
+    )
+    assert "result" in ret.data
+    assert ret.data["result"]
+
+    assert "comment" in ret.data
+    assert ret.data["comment"] == "Added job: job1 to schedule."
+
+    assert "changes" in ret.data
+    assert ret.data["changes"] == {"job1": "added"}
+
+    _expected = """schedule:
+  job1:
+    enabled: true
+    function: test.ping
+    jid_include: true
+    maxrunning: 1
+    name: job1
+    saved: true
+"""
+    ret = salt_cli.run("schedule.list", minion_tgt=proxy_id)
+    assert ret.data == _expected
+
+    # clean out the scheduler
+    salt_cli.run("schedule.purge", minion_tgt=proxy_id)
+
+
+def test_schedule_add_list_all(salt_cli, proxy_ids):
+    """
+    Ensure schedule.add works when targeting a single minion
+    and that the others are not affected.
+    """
+    ret = salt_cli.run(
+        "schedule.add", name="job2", function="test.ping", minion_tgt=proxy_ids[0]
+    )
+    assert "result" in ret.data
+    assert ret.data["result"]
+
+    ret = salt_cli.run("-L", "schedule.list", minion_tgt=",".join(proxy_ids))
+
+    # check every proxy except the first one
+    for _id in proxy_ids[1:]:
+        assert ret.data[_id] == "schedule: {}\n"
+
+    # clean out the scheduler
+    salt_cli.run("-L", "schedule.purge", minion_tgt=",".join(proxy_ids))
diff --git a/tests/pytests/unit/proxy/test_ssh_sample.py b/tests/pytests/unit/proxy/test_ssh_sample.py
new file mode 100644
index 000000000000..58bfb30eb6bd
--- /dev/null
+++ b/tests/pytests/unit/proxy/test_ssh_sample.py
@@ -0,0 +1,269 @@
+"""
+    :codeauthor: Gareth J. Greenaway <gareth@saltstack.com>
+"""
+
+import copy
+import logging
+
+import pytest
+from saltfactories.utils import random_string
+
+import salt.proxy.ssh_sample as ssh_sample_proxy
+from salt.utils.vt import TerminalException
+from tests.support.mock import MagicMock, patch
+
+log = logging.getLogger(__name__)
+
+
+@pytest.fixture(scope="module")
+def proxy_minion_config_module(salt_master_factory):
+    factory = salt_master_factory.salt_proxy_minion_daemon(
+        random_string("proxy-minion-"),
+    )
+    return factory.config
+
+
+@pytest.fixture
+def proxy_minion_config(proxy_minion_config_module):
+
+    minion_config = copy.deepcopy(proxy_minion_config_module)
+    minion_config["proxy"]["proxytype"] = "ssh_sample"
+    minion_config["proxy"]["host"] = "localhost"
+    minion_config["proxy"]["username"] = "username"
+    minion_config["proxy"]["password"] = "password"
+    return minion_config
+
+
+@pytest.fixture
+def configure_loader_modules():
+    return {ssh_sample_proxy: {}}
+
+
+class MockSSHConnection:
+    def __init__(self, *args, **kwargs):
+        return None
+
+    def sendline(self, *args, **kwargs):
+        return "", ""
+
+
+def test_init(proxy_minion_config):
+    """
+    check ssh_sample_proxy init method
+    """
+
+    with patch(
+        "salt.utils.vt_helper.SSHConnection",
+        MagicMock(autospec=True, return_value=MockSSHConnection()),
+    ):
+        with patch.dict(ssh_sample_proxy.__opts__, proxy_minion_config):
+            ssh_sample_proxy.init(proxy_minion_config)
+            assert "server" in ssh_sample_proxy.__context__
+            assert "initialized" in ssh_sample_proxy.__context__
+            assert ssh_sample_proxy.__context__["initialized"]
+
+
+def test_initialized(proxy_minion_config):
+    """
+    check ssh_sample_proxy initialized method
+    """
+
+    with patch(
+        "salt.utils.vt_helper.SSHConnection",
+        MagicMock(autospec=True, return_value=MockSSHConnection()),
+    ):
+        with patch.dict(ssh_sample_proxy.__opts__, proxy_minion_config):
+            ssh_sample_proxy.init(proxy_minion_config)
+
+            ret = ssh_sample_proxy.initialized()
+            assert ret
+
+
+def test_grains(proxy_minion_config):
+    """
+    check ssh_sample_proxy grains method
+    """
+
+    GRAINS_INFO = """{
+  "os": "SshExampleOS",
+  "kernel": "0.0000001",
+  "housecat": "Are you kidding?"
+}
+"""
+    mock_sendline = MagicMock(autospec=True, side_effect=[("", ""), (GRAINS_INFO, "")])
+    with patch.object(MockSSHConnection, "sendline", mock_sendline):
+        with patch(
+            "salt.utils.vt_helper.SSHConnection",
+            MagicMock(autospec=True, return_value=MockSSHConnection()),
+        ):
+            with patch.dict(ssh_sample_proxy.__opts__, proxy_minion_config):
+                ssh_sample_proxy.init(proxy_minion_config)
+
+                ret = ssh_sample_proxy.grains()
+                assert "os" in ret
+                assert "kernel" in ret
+                assert "housecat" in ret
+
+                assert ret["os"] == "SshExampleOS"
+                assert ret["kernel"] == "0.0000001"
+                assert ret["housecat"] == "Are you kidding?"
+
+    # Read from __context__ cache
+    mock_sendline = MagicMock(autospec=True, side_effect=[("", ""), (GRAINS_INFO, "")])
+    mock_context = {
+        "grains_cache": {
+            "os": "SSH-ExampleOS",
+            "kernel": "0.0000002",
+            "dog": "Not kidding.",
+        }
+    }
+
+    with patch.object(MockSSHConnection, "sendline", mock_sendline):
+        with patch(
+            "salt.utils.vt_helper.SSHConnection",
+            MagicMock(autospec=True, return_value=MockSSHConnection()),
+        ):
+            with patch.dict(ssh_sample_proxy.__opts__, proxy_minion_config):
+                with patch.dict(ssh_sample_proxy.__context__, mock_context):
+                    ssh_sample_proxy.init(proxy_minion_config)
+
+                    ret = ssh_sample_proxy.grains()
+                    assert "os" in ret
+                    assert "kernel" in ret
+                    assert "dog" in ret
+
+                    assert ret["os"] == "SSH-ExampleOS"
+                    assert ret["kernel"] == "0.0000002"
+                    assert ret["dog"] == "Not kidding."
+
+
+def test_grains_refresh(proxy_minion_config):
+    """
+    check ssh_sample_proxy grains_refresh method
+    """
+
+    GRAINS_INFO = """{
+  "os": "SshExampleOS",
+  "kernel": "0.0000001",
+  "housecat": "Are you kidding?"
+}
+"""
+
+    mock_sendline = MagicMock(autospec=True, side_effect=[("", ""), (GRAINS_INFO, "")])
+    mock_context = {
+        "grains_cache": {
+            "os": "SSH-ExampleOS",
+            "kernel": "0.0000002",
+            "dog": "Not kidding.",
+        }
+    }
+
+    with patch.object(MockSSHConnection, "sendline", mock_sendline):
+        with patch(
+            "salt.utils.vt_helper.SSHConnection",
+            MagicMock(autospec=True, return_value=MockSSHConnection()),
+        ):
+            with patch.dict(ssh_sample_proxy.__opts__, proxy_minion_config):
+                with patch.dict(ssh_sample_proxy.__context__, mock_context):
+                    ssh_sample_proxy.init(proxy_minion_config)
+
+                    ret = ssh_sample_proxy.grains_refresh()
+                    assert "os" in ret
+                    assert "kernel" in ret
+                    assert "housecat" in ret
+
+                    assert ret["os"] == "SshExampleOS"
+                    assert ret["kernel"] == "0.0000001"
+                    assert ret["housecat"] == "Are you kidding?"
+
+
+def test_fns():
+    """
+    check ssh_sample_proxy fns method
+    """
+
+    ret = ssh_sample_proxy.fns()
+
+    assert "details" in ret
+    assert ret["details"] == (
+        "This key is here because a function in "
+        "grains/ssh_sample.py called fns() here in the proxymodule."
+    )
+
+
+def test_ping(proxy_minion_config):
+    """
+    check ssh_sample_proxy ping method
+    """
+
+    mock_sendline = MagicMock(autospec=True, side_effect=[("", ""), ("", "")])
+    with patch.object(MockSSHConnection, "sendline", mock_sendline):
+        with patch(
+            "salt.utils.vt_helper.SSHConnection",
+            MagicMock(autospec=True, return_value=MockSSHConnection()),
+        ):
+            with patch.dict(ssh_sample_proxy.__opts__, proxy_minion_config):
+                ssh_sample_proxy.init(proxy_minion_config)
+
+                ret = ssh_sample_proxy.ping()
+                assert ret
+
+    mock_sendline = MagicMock(autospec=True, side_effect=[("", ""), TerminalException])
+    with patch.object(MockSSHConnection, "sendline", mock_sendline):
+        with patch(
+            "salt.utils.vt_helper.SSHConnection",
+            MagicMock(autospec=True, return_value=MockSSHConnection()),
+        ):
+            with patch.dict(ssh_sample_proxy.__opts__, proxy_minion_config):
+                ssh_sample_proxy.init(proxy_minion_config)
+                ret = ssh_sample_proxy.ping()
+                assert not ret
+
+
+def test_package_list(proxy_minion_config):
+    """
+    check ssh_sample_proxy package_list method
+    """
+
+    PKG_LIST = """{
+  "coreutils": "1.05"
+}
+"""
+
+    mock_sendline = MagicMock(autospec=True, side_effect=[("", ""), (PKG_LIST, "")])
+    with patch.object(MockSSHConnection, "sendline", mock_sendline):
+        with patch(
+            "salt.utils.vt_helper.SSHConnection",
+            MagicMock(autospec=True, return_value=MockSSHConnection()),
+        ):
+            with patch.dict(ssh_sample_proxy.__opts__, proxy_minion_config):
+                ssh_sample_proxy.init(proxy_minion_config)
+
+                ret = ssh_sample_proxy.package_list()
+                assert ret
+                assert "coreutils" in ret
+                assert ret["coreutils"] == "1.05"
+
+
+def test_package_install(proxy_minion_config):
+    """
+    check ssh_sample_proxy package_list method
+    """
+    PKG_INSTALL = """{
+  "redbull": "1.0"
+}
+"""
+
+    mock_sendline = MagicMock(autospec=True, side_effect=[("", ""), (PKG_INSTALL, "")])
+    with patch.object(MockSSHConnection, "sendline", mock_sendline):
+        with patch(
+            "salt.utils.vt_helper.SSHConnection",
+            MagicMock(autospec=True, return_value=MockSSHConnection()),
+        ):
+            with patch.dict(ssh_sample_proxy.__opts__, proxy_minion_config):
+                ssh_sample_proxy.init(proxy_minion_config)
+
+                ret = ssh_sample_proxy.package_install("redbull")
+                assert ret
+                assert "redbull" in ret
+                assert ret["redbull"] == "1.0"