diff --git a/eth/abc.py b/eth/abc.py
index 97a1318e02..6bc971ffaf 100644
--- a/eth/abc.py
+++ b/eth/abc.py
@@ -1542,17 +1542,28 @@ def get_log_entries(self) -> Tuple[Tuple[bytes, Tuple[int, ...], bytes], ...]:
     #
     # State Transition
     #
+    @classmethod
     @abstractmethod
-    def apply_message(self) -> 'ComputationAPI':
+    def apply_message(
+            cls,
+            state: 'StateAPI',
+            message: MessageAPI,
+            transaction_context: TransactionContextAPI) -> 'ComputationAPI':
         """
-        Execution of a VM message.
+        Execute a VM message. This is where the VM-specific call logic exists.
         """
         ...
 
+    @classmethod
     @abstractmethod
-    def apply_create_message(self) -> 'ComputationAPI':
+    def apply_create_message(
+            cls,
+            state: 'StateAPI',
+            message: MessageAPI,
+            transaction_context: TransactionContextAPI) -> 'ComputationAPI':
         """
-        Execution of a VM message to create a new contract.
+        Execute a VM message to create a new contract. This is where the VM-specific
+        create logic exists.
         """
         ...
 
@@ -1563,7 +1574,14 @@ def apply_computation(cls,
                           message: MessageAPI,
                           transaction_context: TransactionContextAPI) -> 'ComputationAPI':
         """
-        Perform the computation that would be triggered by the VM message.
+        Execute the logic within the message: Either run the precompile, or
+        step through each opcode.  Generally, the only VM-specific logic is for
+        each opcode as it executes.
+
+        This should rarely be called directly, because it will skip over other important
+        VM-specific logic that happens before or after the execution.
+
+        Instead, prefer :meth:`~apply_message` or :meth:`~apply_create_message`.
         """
         ...
 
@@ -2499,7 +2517,17 @@ def execute_bytecode(self,
                          code_address: Address = None) -> ComputationAPI:
         """
         Execute raw bytecode in the context of the current state of
-        the virtual machine.
+        the virtual machine. Note that this skips over some of the logic
+        that would normally happen during a call. Watch out for:
+
+            - value (ether) is *not* transferred
+            - state is *not* rolled back in case of an error
+            - The target account is *not* necessarily created
+            - others...
+
+        For other potential surprises, check the implementation differences
+        between :meth:`ComputationAPI.apply_computation` and
+        :meth:`ComputationAPI.apply_message`. (depending on the VM fork)
         """
         ...
 
diff --git a/eth/vm/base.py b/eth/vm/base.py
index ac8edeca3c..64df6f74a1 100644
--- a/eth/vm/base.py
+++ b/eth/vm/base.py
@@ -216,7 +216,7 @@ def execute_bytecode(self,
         )
 
         # Execute it in the VM
-        return self.state.get_computation(message, transaction_context).apply_computation(
+        return self.state.computation_class.apply_computation(
             self.state,
             message,
             transaction_context,
diff --git a/eth/vm/computation.py b/eth/vm/computation.py
index 38e87b7255..5b601591fb 100644
--- a/eth/vm/computation.py
+++ b/eth/vm/computation.py
@@ -1,6 +1,3 @@
-from abc import (
-    abstractmethod,
-)
 import itertools
 from types import TracebackType
 from typing import (
@@ -369,17 +366,17 @@ def apply_child_computation(self, child_msg: MessageAPI) -> ComputationAPI:
 
     def generate_child_computation(self, child_msg: MessageAPI) -> ComputationAPI:
         if child_msg.is_create:
-            child_computation = self.__class__(
+            child_computation = self.apply_create_message(
                 self.state,
                 child_msg,
                 self.transaction_context,
-            ).apply_create_message()
+            )
         else:
-            child_computation = self.__class__(
+            child_computation = self.apply_message(
                 self.state,
                 child_msg,
                 self.transaction_context,
-            ).apply_message()
+            )
         return child_computation
 
     def add_child_computation(self, child_computation: ComputationAPI) -> None:
@@ -513,14 +510,6 @@ def __exit__(self,
     #
     # State Transition
     #
-    @abstractmethod
-    def apply_message(self) -> ComputationAPI:
-        raise NotImplementedError("Must be implemented by subclasses")
-
-    @abstractmethod
-    def apply_create_message(self) -> ComputationAPI:
-        raise NotImplementedError("Must be implemented by subclasses")
-
     @classmethod
     def apply_computation(cls,
                           state: StateAPI,
diff --git a/eth/vm/forks/frontier/computation.py b/eth/vm/forks/frontier/computation.py
index 5841fa80b9..d34bfd2207 100644
--- a/eth/vm/forks/frontier/computation.py
+++ b/eth/vm/forks/frontier/computation.py
@@ -15,6 +15,9 @@
 )
 from eth.abc import (
     ComputationAPI,
+    MessageAPI,
+    StateAPI,
+    TransactionContextAPI,
 )
 from eth.exceptions import (
     OutOfGas,
@@ -45,47 +48,59 @@ class FrontierComputation(BaseComputation):
     opcodes = FRONTIER_OPCODES
     _precompiles = FRONTIER_PRECOMPILES     # type: ignore # https://github.com/python/mypy/issues/708 # noqa: E501
 
-    def apply_message(self) -> ComputationAPI:
-        snapshot = self.state.snapshot()
+    @classmethod
+    def apply_message(
+            cls,
+            state: StateAPI,
+            message: MessageAPI,
+            transaction_context: TransactionContextAPI) -> ComputationAPI:
 
-        if self.msg.depth > STACK_DEPTH_LIMIT:
+        snapshot = state.snapshot()
+
+        if message.depth > STACK_DEPTH_LIMIT:
             raise StackDepthLimit("Stack depth limit reached")
 
-        if self.msg.should_transfer_value and self.msg.value:
-            sender_balance = self.state.get_balance(self.msg.sender)
+        if message.should_transfer_value and message.value:
+            sender_balance = state.get_balance(message.sender)
 
-            if sender_balance < self.msg.value:
+            if sender_balance < message.value:
                 raise InsufficientFunds(
-                    f"Insufficient funds: {sender_balance} < {self.msg.value}"
+                    f"Insufficient funds: {sender_balance} < {message.value}"
                 )
 
-            self.state.delta_balance(self.msg.sender, -1 * self.msg.value)
-            self.state.delta_balance(self.msg.storage_address, self.msg.value)
+            state.delta_balance(message.sender, -1 * message.value)
+            state.delta_balance(message.storage_address, message.value)
 
-            self.logger.debug2(
+            cls.logger.debug2(
                 "TRANSFERRED: %s from %s -> %s",
-                self.msg.value,
-                encode_hex(self.msg.sender),
-                encode_hex(self.msg.storage_address),
+                message.value,
+                encode_hex(message.sender),
+                encode_hex(message.storage_address),
             )
 
-        self.state.touch_account(self.msg.storage_address)
+        state.touch_account(message.storage_address)
 
-        computation = self.apply_computation(
-            self.state,
-            self.msg,
-            self.transaction_context,
+        computation = cls.apply_computation(
+            state,
+            message,
+            transaction_context,
         )
 
         if computation.is_error:
-            self.state.revert(snapshot)
+            state.revert(snapshot)
         else:
-            self.state.commit(snapshot)
+            state.commit(snapshot)
 
         return computation
 
-    def apply_create_message(self) -> ComputationAPI:
-        computation = self.apply_message()
+    @classmethod
+    def apply_create_message(
+            cls,
+            state: StateAPI,
+            message: MessageAPI,
+            transaction_context: TransactionContextAPI) -> ComputationAPI:
+
+        computation = cls.apply_message(state, message, transaction_context)
 
         if computation.is_error:
             return computation
@@ -102,11 +117,11 @@ def apply_create_message(self) -> ComputationAPI:
                 except OutOfGas:
                     computation.output = b''
                 else:
-                    self.logger.debug2(
+                    cls.logger.debug2(
                         "SETTING CODE: %s -> length: %s | hash: %s",
-                        encode_hex(self.msg.storage_address),
+                        encode_hex(message.storage_address),
                         len(contract_code),
                         encode_hex(keccak(contract_code))
                     )
-                    self.state.set_code(self.msg.storage_address, contract_code)
+                    state.set_code(message.storage_address, contract_code)
             return computation
diff --git a/eth/vm/forks/frontier/state.py b/eth/vm/forks/frontier/state.py
index f3920e4fad..32d10ea741 100644
--- a/eth/vm/forks/frontier/state.py
+++ b/eth/vm/forks/frontier/state.py
@@ -123,14 +123,17 @@ def build_computation(self,
                     encode_hex(message.storage_address),
                 )
             else:
-                computation = self.vm_state.get_computation(
+                computation = self.vm_state.computation_class.apply_create_message(
+                    self.vm_state,
                     message,
                     transaction_context,
-                ).apply_create_message()
+                )
         else:
-            computation = self.vm_state.get_computation(
+            computation = self.vm_state.computation_class.apply_message(
+                self.vm_state,
                 message,
-                transaction_context).apply_message()
+                transaction_context,
+            )
 
         return computation
 
diff --git a/eth/vm/forks/homestead/computation.py b/eth/vm/forks/homestead/computation.py
index a002a9034b..bafa9d9d27 100644
--- a/eth/vm/forks/homestead/computation.py
+++ b/eth/vm/forks/homestead/computation.py
@@ -8,7 +8,12 @@
     encode_hex,
 )
 
-from eth.abc import ComputationAPI
+from eth.abc import (
+    ComputationAPI,
+    MessageAPI,
+    StateAPI,
+    TransactionContextAPI,
+)
 from eth.vm.forks.frontier.computation import (
     FrontierComputation,
 )
@@ -24,13 +29,18 @@ class HomesteadComputation(FrontierComputation):
     # Override
     opcodes = HOMESTEAD_OPCODES
 
-    def apply_create_message(self) -> ComputationAPI:
-        snapshot = self.state.snapshot()
+    @classmethod
+    def apply_create_message(
+            cls,
+            state: StateAPI,
+            message: MessageAPI,
+            transaction_context: TransactionContextAPI) -> ComputationAPI:
+        snapshot = state.snapshot()
 
-        computation = self.apply_message()
+        computation = cls.apply_message(state, message, transaction_context)
 
         if computation.is_error:
-            self.state.revert(snapshot)
+            state.revert(snapshot)
             return computation
         else:
             contract_code = computation.output
@@ -46,18 +56,18 @@ def apply_create_message(self) -> ComputationAPI:
                     # Different from Frontier: reverts state on gas failure while
                     # writing contract code.
                     computation.error = err
-                    self.state.revert(snapshot)
+                    state.revert(snapshot)
                 else:
-                    if self.logger:
-                        self.logger.debug2(
+                    if cls.logger:
+                        cls.logger.debug2(
                             "SETTING CODE: %s -> length: %s | hash: %s",
-                            encode_hex(self.msg.storage_address),
+                            encode_hex(message.storage_address),
                             len(contract_code),
                             encode_hex(keccak(contract_code))
                         )
 
-                    self.state.set_code(self.msg.storage_address, contract_code)
-                    self.state.commit(snapshot)
+                    state.set_code(message.storage_address, contract_code)
+                    state.commit(snapshot)
             else:
-                self.state.commit(snapshot)
+                state.commit(snapshot)
             return computation
diff --git a/eth/vm/forks/spurious_dragon/computation.py b/eth/vm/forks/spurious_dragon/computation.py
index fd4b96100b..85317ba784 100644
--- a/eth/vm/forks/spurious_dragon/computation.py
+++ b/eth/vm/forks/spurious_dragon/computation.py
@@ -4,7 +4,12 @@
 )
 
 from eth import constants
-from eth.abc import ComputationAPI
+from eth.abc import (
+    ComputationAPI,
+    MessageAPI,
+    StateAPI,
+    TransactionContextAPI,
+)
 from eth.exceptions import (
     OutOfGas,
 )
@@ -24,16 +29,22 @@ class SpuriousDragonComputation(HomesteadComputation):
     # Override
     opcodes = SPURIOUS_DRAGON_OPCODES
 
-    def apply_create_message(self) -> ComputationAPI:
-        snapshot = self.state.snapshot()
+    @classmethod
+    def apply_create_message(
+            cls,
+            state: StateAPI,
+            message: MessageAPI,
+            transaction_context: TransactionContextAPI) -> ComputationAPI:
+
+        snapshot = state.snapshot()
 
         # EIP161 nonce incrementation
-        self.state.increment_nonce(self.msg.storage_address)
+        state.increment_nonce(message.storage_address)
 
-        computation = self.apply_message()
+        computation = cls.apply_message(state, message, transaction_context)
 
         if computation.is_error:
-            self.state.revert(snapshot)
+            state.revert(snapshot)
             return computation
         else:
             contract_code = computation.output
@@ -43,7 +54,7 @@ def apply_create_message(self) -> ComputationAPI:
                     f"Contract code size exceeds EIP170 limit of {EIP170_CODE_SIZE_LIMIT}."
                     f"  Got code of size: {len(contract_code)}"
                 )
-                self.state.revert(snapshot)
+                state.revert(snapshot)
             elif contract_code:
                 contract_code_gas_cost = len(contract_code) * constants.GAS_CODEDEPOSIT
                 try:
@@ -55,18 +66,18 @@ def apply_create_message(self) -> ComputationAPI:
                     # Different from Frontier: reverts state on gas failure while
                     # writing contract code.
                     computation.error = err
-                    self.state.revert(snapshot)
+                    state.revert(snapshot)
                 else:
-                    if self.logger:
-                        self.logger.debug2(
+                    if cls.logger:
+                        cls.logger.debug2(
                             "SETTING CODE: %s -> length: %s | hash: %s",
-                            encode_hex(self.msg.storage_address),
+                            encode_hex(message.storage_address),
                             len(contract_code),
                             encode_hex(keccak(contract_code))
                         )
 
-                    self.state.set_code(self.msg.storage_address, contract_code)
-                    self.state.commit(snapshot)
+                    state.set_code(message.storage_address, contract_code)
+                    state.commit(snapshot)
             else:
-                self.state.commit(snapshot)
+                state.commit(snapshot)
             return computation
diff --git a/newsfragments/1921.internal.rst b/newsfragments/1921.internal.rst
new file mode 100644
index 0000000000..2de09b18f2
--- /dev/null
+++ b/newsfragments/1921.internal.rst
@@ -0,0 +1,5 @@
+Fix for creating a duplicate "ghost" Computation that was never used. It didn't
+break anything, but was ineligant and surprising to get extra objects created
+that were mostly useless. This was achieved by changing
+:meth:`ComputationAPI.apply_message` and
+:meth:`ComputationAPI.apply_create_message` to be class methods.
diff --git a/pytest.ini b/pytest.ini
index 96a80ab759..6c2e4eccb4 100644
--- a/pytest.ini
+++ b/pytest.ini
@@ -1,6 +1,9 @@
 [pytest]
-addopts= --showlocals
+addopts= --showlocals --durations 10
 python_paths= .
 xfail_strict=true
 log_format = %(levelname)8s  %(asctime)s  %(filename)20s  %(message)s
 log_date_format = %m-%d %H:%M:%S
+
+[pytest-watch]
+runner= pytest --failed-first --maxfail=1
diff --git a/tests/core/opcodes/test_opcodes.py b/tests/core/opcodes/test_opcodes.py
index 267954a7a5..fa9e4cdebd 100644
--- a/tests/core/opcodes/test_opcodes.py
+++ b/tests/core/opcodes/test_opcodes.py
@@ -828,7 +828,11 @@ def test_sstore(vm_class, code, gas_used, refund, original):
     assert computation.state.get_storage(CANONICAL_ADDRESS_B, 0, from_journal=True) == original
     assert computation.state.get_storage(CANONICAL_ADDRESS_B, 0, from_journal=False) == original
 
-    comp = computation.apply_message()
+    comp = computation.apply_message(
+        computation.state,
+        computation.msg,
+        computation.transaction_context,
+    )
     assert comp.get_gas_refund() == refund
     assert comp.get_gas_used() == gas_used
 
@@ -858,7 +862,11 @@ def test_sstore_limit_2300(gas_supplied, success, gas_used, refund):
     assert computation.state.get_storage(CANONICAL_ADDRESS_B, 0) == original
     computation.state.persist()
 
-    comp = computation.apply_message()
+    comp = computation.apply_message(
+        computation.state,
+        computation.msg,
+        computation.transaction_context,
+    )
     if success and not comp.is_success:
         raise comp._error
     else:
@@ -963,7 +971,11 @@ def test_balance(vm_class, code, expect_exception, expect_gas_used):
     computation.state.set_balance(CANONICAL_ADDRESS_B, sender_balance)
     computation.state.persist()
 
-    comp = computation.apply_message()
+    comp = computation.apply_message(
+        computation.state,
+        computation.msg,
+        computation.transaction_context,
+    )
     if expect_exception:
         assert isinstance(comp.error, expect_exception)
     else:
@@ -1017,7 +1029,11 @@ def test_balance(vm_class, code, expect_exception, expect_gas_used):
 )
 def test_gas_costs(vm_class, code, expect_gas_used):
     computation = setup_computation(vm_class, CANONICAL_ADDRESS_B, code)
-    comp = computation.apply_message()
+    comp = computation.apply_message(
+        computation.state,
+        computation.msg,
+        computation.transaction_context,
+    )
     assert comp.is_success
     assert comp.get_gas_used() == expect_gas_used
 
@@ -1098,7 +1114,11 @@ def test_blake2b_f_compression(vm_class, input_hex, output_hex, expect_exception
         data=to_bytes(hexstr=input_hex),
     )
 
-    comp = computation.apply_message()
+    comp = computation.apply_message(
+        computation.state,
+        computation.msg,
+        computation.transaction_context,
+    )
     if expect_exception:
         assert isinstance(comp.error, expect_exception)
     else:
diff --git a/tests/core/vm/test_base_computation.py b/tests/core/vm/test_base_computation.py
index 7421662497..d07b51e410 100644
--- a/tests/core/vm/test_base_computation.py
+++ b/tests/core/vm/test_base_computation.py
@@ -27,11 +27,13 @@
 
 
 class DummyComputation(BaseComputation):
-    def apply_message(self):
-        return self
+    @classmethod
+    def apply_message(cls, *args):
+        return cls(*args)
 
-    def apply_create_message(self):
-        return self
+    @classmethod
+    def apply_create_message(cls, *args):
+        return cls(*args)
 
 
 class DummyTransactionContext(BaseTransactionContext):
diff --git a/tests/json-fixtures/test_virtual_machine.py b/tests/json-fixtures/test_virtual_machine.py
index da095abc06..8a70e1a3ad 100644
--- a/tests/json-fixtures/test_virtual_machine.py
+++ b/tests/json-fixtures/test_virtual_machine.py
@@ -87,18 +87,18 @@ def fixture(fixture_data):
 #
 # Testing Overrides
 #
-def apply_message_for_testing(self):
+def apply_message_for_testing(cls, *args):
     """
     For VM tests, we don't actually apply messages.
     """
-    return self
+    return cls(*args)
 
 
-def apply_create_message_for_testing(self):
+def apply_create_message_for_testing(cls, *args):
     """
     For VM tests, we don't actually apply messages.
     """
-    return self
+    return cls(*args)
 
 
 def get_block_hash_for_testing(self, block_number):