# Copyright (c) Microsoft Corporation.
# SPDX-License-Identifier: Apache-2.0

# DeepSpeed Team

import torch

from .abstract_accelerator import DeepSpeedAccelerator

# During setup stage torch may not be installed, pass on no torch will
# allow op builder related API to be executed.
try:
    import torch.mps
except ImportError:
    pass


class MPS_Accelerator(DeepSpeedAccelerator):

    def __init__(self):
        self._name = "mps"
        self._communication_backend_name = None

    def is_synchronized_device(self):
        return False

    # Device APIs
    def device_name(self, device_index=None):
        if device_index is None:
            return "mps"
        return "mps:{}".format(device_index)

    def device(self, device_index):
        return torch.device("mps", index=0)

    def set_device(self, device_index):
        return

    def current_device(self):
        return torch.device("mps", index=0)

    def current_device_name(self):
        return "mps:0"

    def device_count(self):
        return 1

    def synchronize(self, device_index=None):
        return torch.mps.synchronize()

    # RNG APIs
    def random(self):
        return torch.random

    def set_rng_state(self, new_state, device_index=None):
        return torch.mps.set_rng_state(new_state)

    def get_rng_state(self, device_index=None):
        return torch.mps.get_rng_state()

    def manual_seed(self, seed):
        return torch.mps.manual_seed(seed)

    def manual_seed_all(self, seed):
        return torch.mps.manual_seed(seed)

    def seed(self):
        return torch.mps.seed()

    def initial_seed(self, seed):
        return

    def default_generator(self, device_index):
        return

    # Streams/Events
    @property
    def Stream(self):
        return None

    def stream(self, stream):
        return None

    def current_stream(self, device_index=None):
        return None

    def default_stream(self, device_index=None):
        return None

    @property
    def Event(self):
        return None

    # Memory management
    def empty_cache(self):
        return torch.mps.empty_cache()

    def memory_allocated(self, device_index=None):
        return torch.mps.current_allocated_memory()

    def max_memory_allocated(self, device_index=None):
        return torch.mps.driver_allocated_memory()

    def set_per_process_memory_fraction(self, fraction):
        return torch.mps.set_per_process_memory_fraction(fraction)

    def reset_max_memory_allocated(self, device_index=None):
        return

    def memory_cached(self, device_index=None):
        return

    def max_memory_cached(self, device_index=None):
        return

    def reset_max_memory_cached(self, device_index=None):
        return

    def memory_stats(self, device_index=None):
        return

    def reset_peak_memory_stats(self, device_index=None):
        return

    def memory_reserved(self, device_index=None):
        return

    def max_memory_reserved(self, device_index=None):
        return

    def total_memory(self, device_index=None):
        return

    def available_memory(self, device_index=None):
        return

    # Data types
    def is_bf16_supported(self):
        return False

    def is_fp16_supported(self):
        return False

    def supported_dtypes(self):
        return [torch.float]

    # Misc
    def amp(self):
        return

    def is_available(self):
        return hasattr(torch.backends, "mps") and torch.backends.mps.is_available()

    def range_push(self, msg):
        return

    def range_pop(self):
        return

    def lazy_call(self, callback):
        return

    def communication_backend_name(self):
        return self._communication_backend_name

    def is_triton_supported(self):
        return False

    # Graph operations
    def create_graph(self):
        return None

    def capture_to_graph(self, graph, pool=None, stream=None):
        from deepspeed.runtime.utils import noop_context
        return noop_context()

    def replay_graph(self, graph):
        return

    # Tensor operations
    @property
    def BFloat16Tensor(self):
        return

    @property
    def ByteTensor(self):
        return

    @property
    def DoubleTensor(self):
        return

    @property
    def FloatTensor(self):
        return

    @property
    def HalfTensor(self):
        return

    @property
    def IntTensor(self):
        return

    @property
    def LongTensor(self):
        return

    def pin_memory(self, tensor, align_bytes=1):
        return tensor.pin_memory()

    def is_pinned(self, tensor):
        return tensor.is_pinned()

    def on_accelerator(self, tensor):
        device_str = str(tensor.device)
        if device_str.startswith("mps"):
            return True
        else:
            return False

    def op_builder_dir(self):
        try:
            # is op_builder from deepspeed or a 3p version? this should only succeed if it's deepspeed
            # if successful this also means we're doing a local install and not JIT compile path
            from op_builder import __deepspeed__  # noqa: F401 # type: ignore

            return "op_builder"
        except ImportError:
            return "deepspeed.ops.op_builder"

    # create an instance of op builder, specified by class_name
    def create_op_builder(self, op_name):
        builder_class = self.get_op_builder(op_name)
        if builder_class is not None:
            return builder_class()
        return None

    # return an op builder class, specified by class_name
    def get_op_builder(self, class_name):
        from deepspeed.ops.op_builder.cpu import NotImplementedBuilder

        return NotImplementedBuilder

    def build_extension(self):
        from torch.utils.cpp_extension import BuildExtension

        return BuildExtension

    def export_envs(self):
        return []

accelerator/mps_accelerator.py
# Copyright (c) Microsoft Corporation.
# SPDX-License-Identifier: Apache-2.0

# DeepSpeed Team

import torch

from .abstract_accelerator import DeepSpeedAccelerator

# During setup stage torch may not be installed, pass on no torch will
# allow op builder related API to be executed.
try:
    import torch.mps
except ImportError:
    pass


class MPS_Accelerator(DeepSpeedAccelerator):

    def __init__(self):
        self._name = "mps"
        self._communication_backend_name = None

    def is_synchronized_device(self):
        return False

    def use_host_timers(self):
        return self.is_synchronized_device()

    def resolves_data_dependency(self):
        return self.is_synchronized_device()

    def handles_memory_backpressure(self):
        return self.is_synchronized_device()

    # Device APIs
    def device_name(self, device_index=None):
        if device_index is None:
            return "mps"
        return "mps:{}".format(device_index)

    def device(self, device_index):
        return torch.device("mps", index=0)

    def set_device(self, device_index):
        return

    def current_device(self):
        return torch.device("mps", index=0)

    def current_device_name(self):
        return "mps:0"

    def device_count(self):
        return 1

    def synchronize(self, device_index=None):
        return torch.mps.synchronize()

    # RNG APIs
    def random(self):
        return torch.random

    def set_rng_state(self, new_state, device_index=None):
        return torch.mps.set_rng_state(new_state)

    def get_rng_state(self, device_index=None):
        return torch.mps.get_rng_state()

    def manual_seed(self, seed):
        return torch.mps.manual_seed(seed)

    def manual_seed_all(self, seed):
        return torch.mps.manual_seed(seed)

    def seed(self):
        return torch.mps.seed()

    def initial_seed(self, seed):
        return

    def default_generator(self, device_index):
        return

    # Streams/Events
    @property
    def Stream(self):
        return None

    def stream(self, stream):
        return None

    def current_stream(self, device_index=None):
        return None

    def default_stream(self, device_index=None):
        return None

    @property
    def Event(self):
        return None

    # Memory management
    def empty_cache(self):
        return torch.mps.empty_cache()

    def memory_allocated(self, device_index=None):
        return torch.mps.current_allocated_memory()

    def max_memory_allocated(self, device_index=None):
        return torch.mps.driver_allocated_memory()

    def set_per_process_memory_fraction(self, fraction):
        return torch.mps.set_per_process_memory_fraction(fraction)

    def reset_max_memory_allocated(self, device_index=None):
        return

    def memory_cached(self, device_index=None):
        return

    def max_memory_cached(self, device_index=None):
        return

    def reset_max_memory_cached(self, device_index=None):
        return

    def memory_stats(self, device_index=None):
        return

    def reset_peak_memory_stats(self, device_index=None):
        return

    def memory_reserved(self, device_index=None):
        return

    def max_memory_reserved(self, device_index=None):
        return

    def total_memory(self, device_index=None):
        return

    def available_memory(self, device_index=None):
        return

    # Data types
    def is_bf16_supported(self):
        return False

    def is_fp16_supported(self):
        return False

    def supported_dtypes(self):
        return [torch.float]

    # Misc
    def amp(self):
        return

    def is_available(self):
        return hasattr(torch.backends, "mps") and torch.backends.mps.is_available()

    def range_push(self, msg):
        return

    def range_pop(self):
        return

    def lazy_call(self, callback):
        return

    def communication_backend_name(self):
        return self._communication_backend_name

    def is_triton_supported(self):
        return False

    # Graph operations
    def create_graph(self):
        return None

    def capture_to_graph(self, graph, pool=None, stream=None):
        from deepspeed.runtime.utils import noop_context
        return noop_context()

    def replay_graph(self, graph):
        return

    # Tensor operations
    @property
    def BFloat16Tensor(self):
        return

    @property
    def ByteTensor(self):
        return

    @property
    def DoubleTensor(self):
        return

    @property
    def FloatTensor(self):
        return

    @property
    def HalfTensor(self):
        return

    @property
    def IntTensor(self):
        return

    @property
    def LongTensor(self):
        return

    def pin_memory(self, tensor, align_bytes=1):
        return tensor.pin_memory()

    def is_pinned(self, tensor):
        return tensor.is_pinned()

    def on_accelerator(self, tensor):
        device_str = str(tensor.device)
        if device_str.startswith("mps"):
            return True
        else:
            return False

    def op_builder_dir(self):
        try:
            # is op_builder from deepspeed or a 3p version? this should only succeed if it's deepspeed
            # if successful this also means we're doing a local install and not JIT compile path
            from op_builder import __deepspeed__  # noqa: F401 # type: ignore

            return "op_builder"
        except ImportError:
            return "deepspeed.ops.op_builder"

    # create an instance of op builder, specified by class_name
    def create_op_builder(self, op_name):
        builder_class = self.get_op_builder(op_name)
        if builder_class is not None:
            return builder_class()
        return None

    # return an op builder class, specified by class_name
    def get_op_builder(self, class_name):
        from deepspeed.ops.op_builder.cpu import NotImplementedBuilder

        return NotImplementedBuilder

    def build_extension(self):
        from torch.utils.cpp_extension import BuildExtension

        return BuildExtension

    def export_envs(self):
        return []

accelerator/mps_accelerator.py
diff --git a/accelerator/mps_accelerator.py b/accelerator/mps_accelerator.py
index f6303cf9890f..972b33caece1 100644
--- a/accelerator/mps_accelerator.py
+++ b/accelerator/mps_accelerator.py
@@ -24,6 +24,15 @@ def __init__(self):
     def is_synchronized_device(self):
         return False
 
+    def use_host_timers(self):
+        return self.is_synchronized_device()
+
+    def resolves_data_dependency(self):
+        return self.is_synchronized_device()
+
+    def handles_memory_backpressure(self):
+        return self.is_synchronized_device()
+
     # Device APIs
     def device_name(self, device_index=None):
         if device_index is None:
diff --git a/tests/unit/accelerator/test_accelerator.py b/tests/unit/accelerator/test_accelerator.py
new file mode 100644
index 000000000000..964cf2b24f4e
--- /dev/null
+++ b/tests/unit/accelerator/test_accelerator.py
@@ -0,0 +1,59 @@
+# Copyright (c) Microsoft Corporation.
+# SPDX-License-Identifier: Apache-2.0
+
+# DeepSpeed Team
+
+import pytest
+
+import os
+import sys
+import importlib
+import re
+
+import deepspeed
+
+DS_ACCEL_PATH = "deepspeed.accelerator"
+IGNORE_FILES = ["abstract_accelerator.py", "real_accelerator.py"]
+
+
+@pytest.fixture
+def accel_class_name(module_name):
+    class_list = []
+    mocked_modules = []
+
+    # Get the accelerator class name for a given module
+    while True:
+        try:
+            module = importlib.import_module(module_name)
+            break
+        except ModuleNotFoundError as e:
+            # If the environment is missing a module, mock it so we can still
+            # test importing the accelerator class
+            missing_module = re.search(r"\'(.*)\'", e.msg).group().strip("'")
+            sys.modules[missing_module] = lambda x: None
+            mocked_modules.append(missing_module)
+    for name in dir(module):
+        if name.endswith("_Accelerator"):
+            class_list.append(name)
+
+    assert len(class_list) == 1, f"Multiple accelerator classes found in {module_name}"
+
+    yield class_list[0]
+
+    # Clean up mocked modules so as to not impact other tests
+    for module in mocked_modules:
+        del sys.modules[module]
+
+
+@pytest.mark.parametrize(
+    "module_name",
+    [
+        DS_ACCEL_PATH + "." + f.rstrip(".py") for f in os.listdir(deepspeed.accelerator.__path__[0])
+        if f.endswith("_accelerator.py") and f not in IGNORE_FILES
+    ],
+)
+def test_abstract_methods_defined(module_name, accel_class_name):
+    module = importlib.import_module(module_name)
+    accel_class = getattr(module, accel_class_name)
+    accel_class.__init__ = lambda self: None
+    _ = accel_class()

[BUG] TypeError: Can't instantiate abstract class MPS_Accelerator with abstract methods
**Describe the bug**
TypeError: Can't instantiate abstract class MPS_Accelerator with abstract methods handles_memory_backpressure, resolves_data_dependency, use_host_timers

**To Reproduce**
Steps to reproduce the behavior:
1. git clone https://github.com/OpenBMB/MiniCPM.git
2. follow setup step.
3. run `!bash lora_finetune.sh` via `lora_finetune.ipynb`

**Expected behavior**
runnable

**ds_report output**
Please run `ds_report` to give us details about your setup.

**Screenshots**
```
    class CudaEventTimer(object):
  File "/Users/katopz/git/MiniCPM/finetune/.venv/lib/python3.11/site-packages/deepspeed/utils/timer.py", line 33, in CudaEventTimer
    def __init__(self, start_event: get_accelerator().Event, end_event: get_accelerator().Event):
                                    ^^^^^^^^^^^^^^^^^
  File "/Users/katopz/git/MiniCPM/finetune/.venv/lib/python3.11/site-packages/deepspeed/accelerator/real_accelerator.py", line 184, in get_accelerator
    ds_accelerator = MPS_Accelerator()
                     ^^^^^^^^^^^^^^^^^
TypeError: Can't instantiate abstract class MPS_Accelerator with abstract methods handles_memory_backpressure, resolves_data_dependency, use_host_timers
```

**System info (please complete the following information):**
 - OS: macOS 14.2.1 (23C71)
 - metal
 - Python 3.11.7

**Launcher context**
deepspeed

**Docker context**
no

**Additional context**


