diff --git a/Changelog.md b/Changelog.md
index c907e248..e0886502 100644
--- a/Changelog.md
+++ b/Changelog.md
@@ -5,6 +5,7 @@ All notable changes to this project will be documented here.
 - Update python, pyta and jupyter testers to allow a requirements file (#580)
 - Update R tester to allow a renv.lock file (#581)
 - Improve display of Python package installation errors when creating environment (#585)
+- Change rlimit resource settings to apply each worker individually (#587) 
 
 ## [v2.6.0]
 - Update python versions in docker file (#568)
diff --git a/server/autotest_server/__init__.py b/server/autotest_server/__init__.py
index cf349518..a730cdb9 100644
--- a/server/autotest_server/__init__.py
+++ b/server/autotest_server/__init__.py
@@ -19,7 +19,13 @@
 from types import TracebackType
 
 from .config import config
-from .utils import loads_partial_json, set_rlimits_before_test, extract_zip_stream, recursive_iglob, copy_tree
+from .utils import (
+    loads_partial_json,
+    get_resource_settings,
+    extract_zip_stream,
+    recursive_iglob,
+    copy_tree,
+)
 
 DEFAULT_ENV_DIR = "defaultvenv"
 TEST_SCRIPT_DIR = os.path.join(config["workspace"], "scripts")
@@ -103,7 +109,7 @@ def _create_test_script_command(tester_type: str) -> str:
         f'sys.path.append("{os.path.dirname(os.path.abspath(__file__))}")',
         import_line,
         "from testers.specs import TestSpecs",
-        "Tester(specs=TestSpecs.from_json(sys.stdin.read())).run()",
+        f"Tester(resource_settings={get_resource_settings(config)}, specs=TestSpecs.from_json(sys.stdin.read())).run()",
     ]
     python_str = "; ".join(python_lines)
     return f"\"${{PYTHON}}\" -c '{python_str}'"
@@ -221,7 +227,6 @@ def _run_test_specs(
                         stdout=subprocess.PIPE,
                         stderr=subprocess.PIPE,
                         stdin=subprocess.PIPE,
-                        preexec_fn=set_rlimits_before_test,
                         universal_newlines=True,
                         env={**os.environ, **env_vars, **env},
                         executable="/bin/bash",
diff --git a/server/autotest_server/testers/custom/custom_tester.py b/server/autotest_server/testers/custom/custom_tester.py
index 40aaa6d3..7e879d3d 100644
--- a/server/autotest_server/testers/custom/custom_tester.py
+++ b/server/autotest_server/testers/custom/custom_tester.py
@@ -4,9 +4,9 @@
 
 
 class CustomTester(Tester):
-    def __init__(self, specs: TestSpecs) -> None:
+    def __init__(self, specs: TestSpecs, resource_settings: list[tuple[int, tuple[int, int]]] | None = None) -> None:
         """Initialize a CustomTester"""
-        super().__init__(specs, test_class=None)
+        super().__init__(specs, test_class=None, resource_settings=resource_settings)
 
     @Tester.run_decorator
     def run(self) -> None:
diff --git a/server/autotest_server/testers/haskell/haskell_tester.py b/server/autotest_server/testers/haskell/haskell_tester.py
index 9c4a9d78..32b622d7 100644
--- a/server/autotest_server/testers/haskell/haskell_tester.py
+++ b/server/autotest_server/testers/haskell/haskell_tester.py
@@ -53,13 +53,14 @@ def __init__(
         self,
         specs: TestSpecs,
         test_class: Type[HaskellTest] = HaskellTest,
+        resource_settings: list[tuple[int, tuple[int, int]]] | None = None,
     ) -> None:
         """
         Initialize a Haskell tester using the specifications in specs.
 
         This tester will create tests of type test_class.
         """
-        super().__init__(specs, test_class)
+        super().__init__(specs, test_class, resource_settings=resource_settings)
 
     def _test_run_flags(self, test_file: str) -> List[str]:
         """
diff --git a/server/autotest_server/testers/java/java_tester.py b/server/autotest_server/testers/java/java_tester.py
index 247fd919..3479aaf4 100644
--- a/server/autotest_server/testers/java/java_tester.py
+++ b/server/autotest_server/testers/java/java_tester.py
@@ -34,13 +34,18 @@ class JavaTester(Tester):
     JUNIT_JUPITER_RESULT = "TEST-junit-jupiter.xml"
     JUNIT_VINTAGE_RESULT = "TEST-junit-vintage.xml"
 
-    def __init__(self, specs: TestSpecs, test_class: Type[JavaTest] = JavaTest) -> None:
+    def __init__(
+        self,
+        specs: TestSpecs,
+        test_class: Type[JavaTest] = JavaTest,
+        resource_settings: list[tuple[int, tuple[int, int]]] | None = None,
+    ) -> None:
         """
         Initialize a Java tester using the specifications in specs.
 
         This tester will create tests of type test_class.
         """
-        super().__init__(specs, test_class)
+        super().__init__(specs, test_class, resource_settings=resource_settings)
         classpath = self.specs.get("test_data", "classpath", default=".") or "."
         self.java_classpath = ":".join(self._parse_file_paths(classpath))
         self.out_dir = tempfile.TemporaryDirectory(dir=os.getcwd())
diff --git a/server/autotest_server/testers/jupyter/jupyter_tester.py b/server/autotest_server/testers/jupyter/jupyter_tester.py
index 969d774d..765dbda2 100644
--- a/server/autotest_server/testers/jupyter/jupyter_tester.py
+++ b/server/autotest_server/testers/jupyter/jupyter_tester.py
@@ -58,13 +58,14 @@ def __init__(
         self,
         specs: TestSpecs,
         test_class: Type[JupyterTest] = JupyterTest,
+        resource_settings: list[tuple[int, tuple[int, int]]] | None = None,
     ):
         """
         Initialize a jupyter tester using the specifications in specs.
 
         This tester will create tests of type test_class.
         """
-        super().__init__(specs, test_class)
+        super().__init__(specs, test_class, resource_settings=resource_settings)
 
     @staticmethod
     def _run_jupyter_tests(test_file: str) -> List[Dict]:
diff --git a/server/autotest_server/testers/py/py_tester.py b/server/autotest_server/testers/py/py_tester.py
index 88c57129..a6bd2232 100644
--- a/server/autotest_server/testers/py/py_tester.py
+++ b/server/autotest_server/testers/py/py_tester.py
@@ -162,13 +162,14 @@ def __init__(
         self,
         specs: TestSpecs,
         test_class: Type[PyTest] = PyTest,
+        resource_settings: list[tuple[int, tuple[int, int]]] | None = None,
     ):
         """
         Initialize a python tester using the specifications in specs.
 
         This tester will create tests of type test_class.
         """
-        super().__init__(specs, test_class)
+        super().__init__(specs, test_class, resource_settings=resource_settings)
 
     @staticmethod
     def _load_unittest_tests(test_file: str) -> unittest.TestSuite:
diff --git a/server/autotest_server/testers/pyta/pyta_tester.py b/server/autotest_server/testers/pyta/pyta_tester.py
index 767f9f57..6bb15195 100644
--- a/server/autotest_server/testers/pyta/pyta_tester.py
+++ b/server/autotest_server/testers/pyta/pyta_tester.py
@@ -102,13 +102,18 @@ def run(self) -> str:
 class PytaTester(Tester):
     test_class: Type[PytaTest]
 
-    def __init__(self, specs: TestSpecs, test_class: Type[PytaTest] = PytaTest):
+    def __init__(
+        self,
+        specs: TestSpecs,
+        test_class: Type[PytaTest] = PytaTest,
+        resource_settings: list[tuple[int, tuple[int, int]]] | None = None,
+    ):
         """
         Initialize a Python TA tester using the specifications in specs.
 
         This tester will create tests of type test_class.
         """
-        super().__init__(specs, test_class)
+        super().__init__(specs, test_class, resource_settings=resource_settings)
         self.upload_annotations = self.specs.get("test_data", "upload_annotations")
         self.pyta_config = self.update_pyta_config()
         self.annotations = []
diff --git a/server/autotest_server/testers/r/r_tester.py b/server/autotest_server/testers/r/r_tester.py
index a5f99c3b..dfd57602 100644
--- a/server/autotest_server/testers/r/r_tester.py
+++ b/server/autotest_server/testers/r/r_tester.py
@@ -64,13 +64,14 @@ def __init__(
         self,
         specs: TestSpecs,
         test_class: Type[RTest] = RTest,
+        resource_settings: list[tuple[int, tuple[int, int]]] | None = None,
     ) -> None:
         """
         Initialize a R tester using the specifications in specs.
 
         This tester will create tests of type test_class.
         """
-        super().__init__(specs, test_class)
+        super().__init__(specs, test_class, resource_settings=resource_settings)
 
     def run_r_tests(self) -> Dict[str, List[Dict[str, Union[int, str]]]]:
         """
diff --git a/server/autotest_server/testers/racket/racket_tester.py b/server/autotest_server/testers/racket/racket_tester.py
index fc844923..816f783d 100644
--- a/server/autotest_server/testers/racket/racket_tester.py
+++ b/server/autotest_server/testers/racket/racket_tester.py
@@ -38,13 +38,18 @@ def run(self) -> str:
 class RacketTester(Tester):
     ERROR_MSGS = {"bad_json": "Unable to parse test results: {}"}
 
-    def __init__(self, specs, test_class: Type[RacketTest] = RacketTest) -> None:
+    def __init__(
+        self,
+        specs,
+        test_class: Type[RacketTest] = RacketTest,
+        resource_settings: list[tuple[int, tuple[int, int]]] | None = None,
+    ) -> None:
         """
         Initialize a racket tester using the specifications in specs.
 
         This tester will create tests of type test_class.
         """
-        super().__init__(specs, test_class)
+        super().__init__(specs, test_class, resource_settings=resource_settings)
 
     def run_racket_test(self) -> Dict[str, str]:
         """
diff --git a/server/autotest_server/testers/tester.py b/server/autotest_server/testers/tester.py
index abe555b2..f48854c5 100644
--- a/server/autotest_server/testers/tester.py
+++ b/server/autotest_server/testers/tester.py
@@ -4,6 +4,7 @@
 from typing import Optional, Callable, Any, Type, Dict, List
 from .specs import TestSpecs
 import traceback
+import resource
 
 
 class TestError(Exception):
@@ -230,10 +231,16 @@ def __init__(
         self,
         specs: TestSpecs,
         test_class: Optional[Type[Test]] = Test,
+        resource_settings: list[tuple[int, tuple[int, int]]] | None = None,
     ) -> None:
         self.specs = specs
         self.test_class = test_class
 
+        if resource_settings is None:
+            self.resource_settings = []
+        else:
+            self.resource_settings = resource_settings
+
     @staticmethod
     def error_all(message: str, points_total: int = 0, expected: bool = False) -> str:
         """
@@ -257,6 +264,7 @@ def before_tester_run(self) -> None:
         Callback invoked before running this tester.
         Use this for tester initialization steps that can fail, rather than using __init__.
         """
+        self.set_resource_limits(self.resource_settings)
 
     def after_tester_run(self) -> None:
         """
@@ -264,6 +272,12 @@ def after_tester_run(self) -> None:
         Use this for tester cleanup steps that should always be executed, regardless of errors.
         """
 
+    @staticmethod
+    def set_resource_limits(resource_settings: list[tuple[int, tuple[int, int]]]) -> None:
+        """Sets system resource limits using the `resource` package."""
+        for resource_name, rlimit in resource_settings:
+            resource.setrlimit(resource_name, rlimit)
+
     @staticmethod
     def run_decorator(run_func: Callable) -> Callable:
         """
diff --git a/server/autotest_server/tests/test_rlimit.py b/server/autotest_server/tests/test_rlimit.py
new file mode 100644
index 00000000..61375627
--- /dev/null
+++ b/server/autotest_server/tests/test_rlimit.py
@@ -0,0 +1,81 @@
+import unittest
+from unittest.mock import patch, MagicMock
+import resource
+
+from ..config import _Config
+from ..utils import validate_rlimit, get_resource_settings
+
+
+class TestValidateRlimit(unittest.TestCase):
+    def test_normal_limits(self):
+        """Test validate_rlimit with normal positive values."""
+        self.assertEqual(validate_rlimit(100, 200, 150, 250), (100, 200))
+        self.assertEqual(validate_rlimit(200, 300, 100, 250), (100, 250))
+
+    def test_soft_limit_exceeding_hard_limit(self):
+        """Test validate_rlimit where soft limit would exceed hard limit."""
+        self.assertEqual(validate_rlimit(500, 400, 300, 350), (300, 350))
+
+    def test_infinity_values(self):
+        """Test validate_rlimit with -1 (resource.RLIM_INFINITY) values."""
+        self.assertEqual(validate_rlimit(-1, 200, 100, 150), (100, 150))
+        self.assertEqual(validate_rlimit(100, -1, 150, 200), (100, 200))
+        self.assertEqual(validate_rlimit(-1, -1, 100, 200), (100, 200))
+        self.assertEqual(validate_rlimit(100, 200, -1, 150), (100, 150))
+        self.assertEqual(validate_rlimit(100, 200, 150, -1), (100, 200))
+        self.assertEqual(validate_rlimit(100, 200, -1, -1), (100, 200))
+
+    def test_both_negative(self):
+        """Test validate_rlimit where both config and current are negative."""
+        self.assertEqual(validate_rlimit(-1, -1, -1, -1), (-1, -1))
+
+    def test_mixed_negative_cases(self):
+        """Test validate_rlimit with various mixed cases with negative values."""
+        self.assertEqual(validate_rlimit(-1, 200, -1, 300), (-1, 200))
+        self.assertEqual(validate_rlimit(100, -1, -1, -1), (100, -1))
+
+
+class TestGetResourceSettings(unittest.TestCase):
+    @patch("resource.getrlimit")
+    def test_empty_config(self, _):
+        """Test get_resource_settings with an empty config."""
+        config = _Config()
+        config.get = MagicMock(return_value={})
+
+        self.assertEqual(get_resource_settings(config), [])
+
+    @patch("resource.getrlimit")
+    def test_with_config_values(self, mock_getrlimit):
+        """Test get_resource_settings with config containing values."""
+        config = _Config()
+        rlimit_settings = {"nofile": (1024, 2048), "nproc": (30, 60)}
+
+        # Setup config.get to return our rlimit_settings when called with "rlimit_settings"
+        config.get = lambda key, default=None: rlimit_settings if key == "rlimit_settings" else default
+
+        # Setup mock for resource.getrlimit to return different values
+        mock_getrlimit.side_effect = lambda limit: {
+            resource.RLIMIT_NOFILE: (512, 1024),
+            resource.RLIMIT_NPROC: (60, 90),
+        }[limit]
+
+        expected = [(resource.RLIMIT_NOFILE, (512, 1024)), (resource.RLIMIT_NPROC, (30, 60))]
+
+        self.assertEqual(get_resource_settings(config), expected)
+
+    @patch("resource.getrlimit")
+    def test_with_infinity_values(self, mock_getrlimit):
+        """Test get_resource_settings with some infinity (-1) values in the mix."""
+        config = _Config()
+        rlimit_settings = {"nofile": (1024, -1), "nproc": (-1, 60)}
+
+        config.get = lambda key, default=None: rlimit_settings if key == "rlimit_settings" else default
+
+        mock_getrlimit.side_effect = lambda limit: {
+            resource.RLIMIT_NOFILE: (512, 1024),
+            resource.RLIMIT_NPROC: (60, 90),
+        }[limit]
+
+        expected = [(resource.RLIMIT_NOFILE, (512, 1024)), (resource.RLIMIT_NPROC, (60, 60))]
+
+        self.assertEqual(get_resource_settings(config), expected)
diff --git a/server/autotest_server/tests/test_tester.py b/server/autotest_server/tests/test_tester.py
new file mode 100644
index 00000000..82f8fe7c
--- /dev/null
+++ b/server/autotest_server/tests/test_tester.py
@@ -0,0 +1,60 @@
+import unittest
+from unittest.mock import patch, call
+import resource
+from typing import Type
+
+from ..testers.specs import TestSpecs
+from ..testers.tester import Tester, Test
+
+
+class MockTester(Tester):
+    def __init__(
+        self,
+        specs: TestSpecs,
+        test_class: Type[Test] | None = Test,
+        resource_settings: list[tuple[int, tuple[int, int]]] | None = None,
+    ) -> None:
+        super().__init__(specs, test_class, resource_settings)
+
+    def run(self) -> None:
+        pass
+
+
+class TestResourceLimits(unittest.TestCase):
+
+    @patch("resource.setrlimit")
+    def test_set_resource_limits_single_limit(self, mock_setrlimit):
+        """Test setting a single resource limit."""
+        # Arrange
+        tester = MockTester(specs=TestSpecs(), resource_settings=[(resource.RLIMIT_CPU, (10, 20))])
+
+        # Act
+        tester.set_resource_limits(tester.resource_settings)
+
+        # Assert
+        mock_setrlimit.assert_called_once_with(resource.RLIMIT_CPU, (10, 20))
+
+    @patch("resource.setrlimit")
+    def test_set_resource_limits_multiple_limits(self, mock_setrlimit):
+        """Test setting multiple resource limits."""
+        # Arrange
+        tester = MockTester(
+            specs=TestSpecs(),
+            resource_settings=[
+                (resource.RLIMIT_CPU, (10, 20)),
+                (resource.RLIMIT_NOFILE, (1024, 2048)),
+                (resource.RLIMIT_AS, (1024 * 1024 * 100, 1024 * 1024 * 200)),
+            ],
+        )
+
+        # Act
+        tester.set_resource_limits(tester.resource_settings)
+
+        # Assert
+        expected_calls = [
+            call(resource.RLIMIT_CPU, (10, 20)),
+            call(resource.RLIMIT_NOFILE, (1024, 2048)),
+            call(resource.RLIMIT_AS, (1024 * 1024 * 100, 1024 * 1024 * 200)),
+        ]
+        mock_setrlimit.assert_has_calls(expected_calls, any_order=False)
+        self.assertEqual(mock_setrlimit.call_count, 3)
diff --git a/server/autotest_server/utils.py b/server/autotest_server/utils.py
index da2dcdef..d947780b 100644
--- a/server/autotest_server/utils.py
+++ b/server/autotest_server/utils.py
@@ -5,9 +5,7 @@
 import shutil
 from io import BytesIO
 from typing import Type, Optional, Tuple, List, Generator
-from .config import config
-
-RLIMIT_ADJUSTMENTS = {"nproc": 10}
+from .config import _Config
 
 
 def loads_partial_json(json_string: str, expected_type: Optional[Type] = None) -> Tuple[List, bool]:
@@ -45,31 +43,40 @@ def _rlimit_str2int(rlimit_string):
     return getattr(resource, f"RLIMIT_{rlimit_string.upper()}")
 
 
-def set_rlimits_before_test() -> None:
-    """
-    Sets rlimit settings specified in config file
-    This function ensures that for specific limits (defined in RLIMIT_ADJUSTMENTS),
-    there are at least n=RLIMIT_ADJUSTMENTS[limit] resources available for cleanup
-    processes that are not available for test processes.  This ensures that cleanup
-    processes will always be able to run.
+def validate_rlimit(config_soft: int, config_hard: int, curr_soft: int, curr_hard: int) -> tuple[int, int]:
+    """Validates and adjusts resource limits based on configured and current values.
+
+    Returns a tuple containing the validated (soft, hard) limit values. This implementation treats the current soft
+    limit as an upper bound on the config soft limit and will clamp it.
     """
-    for limit_str in config.get("rlimit_settings", {}).keys() | RLIMIT_ADJUSTMENTS.keys():
-        limit = _rlimit_str2int(limit_str)
-        config_soft, config_hard = config.get("rlimit_settings", {}).get(limit_str, resource.getrlimit(limit))
-        curr_soft, curr_hard = resource.getrlimit(limit)
-        # account for the fact that resource.RLIM_INFINITY == -1
-        soft, hard = min(curr_soft, config_soft), min(curr_hard, config_hard)
-        if soft < 0:
-            soft = max(curr_soft, config_soft)
-        if hard < 0:
-            hard = max(curr_hard, config_hard)
-        # reduce the hard limit so that cleanup scripts will have at least adj more resources to use.
-        adj = RLIMIT_ADJUSTMENTS.get(limit_str, 0)
-        if hard >= adj:
-            hard -= adj
-        # make sure the soft limit doesn't exceed the hard limit
+    # account for the fact that resource.RLIM_INFINITY == -1
+    soft, hard = min(curr_soft, config_soft), min(curr_hard, config_hard)
+    if soft < 0:
+        soft = max(curr_soft, config_soft)
+    if hard < 0:
+        hard = max(curr_hard, config_hard)
+    # make sure the soft limit doesn't exceed the hard limit, but keep in mind that -1 is resource.RLIM_INFINITY
+    if hard != -1:
         soft = min(hard, soft)
-        resource.setrlimit(limit, (soft, hard))
+
+    return soft, hard
+
+
+def get_resource_settings(config: _Config) -> list[tuple[int, tuple[int, int]]]:
+    """Returns rlimit settings specified in config file."""
+    resource_settings = []
+
+    for limit_str, rlimit in config.get("rlimit_settings", {}).items():
+        limit = _rlimit_str2int(limit_str)
+
+        rlimit = validate_rlimit(
+            *rlimit,
+            *resource.getrlimit(limit),
+        )
+
+        resource_settings.append((limit, rlimit))
+
+    return resource_settings
 
 
 def extract_zip_stream(zip_byte_stream: bytes, destination: str) -> None: