diff --git a/Cargo.lock b/Cargo.lock
index 17d3685884b18..521e102b98ced 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -91,6 +91,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2c99f64d1e06488f620f932677e24bc6e2897582980441ae90a671415bd7ec2f"
 dependencies = [
  "cfg-if",
+ "getrandom 0.2.7",
  "once_cell",
  "version_check",
 ]
@@ -155,9 +156,11 @@ dependencies = [
  "aptos-config",
  "aptos-crypto",
  "aptos-db-tool",
+ "aptos-debugger",
  "aptos-faucet-core",
  "aptos-framework",
  "aptos-gas",
+ "aptos-gas-profiling",
  "aptos-genesis",
  "aptos-github-client",
  "aptos-global-constants",
@@ -906,6 +909,7 @@ dependencies = [
  "anyhow",
  "aptos-crypto",
  "aptos-gas",
+ "aptos-gas-profiling",
  "aptos-logger",
  "aptos-resource-viewer",
  "aptos-rest-client",
@@ -913,6 +917,7 @@ dependencies = [
  "aptos-types",
  "aptos-validator-interface",
  "aptos-vm",
+ "aptos-vm-logging",
  "clap 3.2.23",
  "move-binary-format",
  "move-cli",
@@ -1385,6 +1390,22 @@ dependencies = [
  "move-core-types",
 ]
 
+[[package]]
+name = "aptos-gas-profiling"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "aptos-framework",
+ "aptos-gas",
+ "aptos-package-builder",
+ "aptos-types",
+ "inferno",
+ "move-binary-format",
+ "move-core-types",
+ "move-vm-types",
+ "regex",
+]
+
 [[package]]
 name = "aptos-genesis"
 version = "0.1.0"
@@ -3633,7 +3654,7 @@ version = "0.2.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d9b39be18770d11421cdb1b9947a45dd3f37e93092cbf377614828a319d5fee8"
 dependencies = [
- "hermit-abi",
+ "hermit-abi 0.1.19",
  "libc",
  "winapi 0.3.9",
 ]
@@ -3664,7 +3685,7 @@ checksum = "c9e3356844c4d6a6d6467b8da2cffb4a2820be256f50a3a386c9d152bab31043"
 dependencies = [
  "async-trait",
  "axum-core 0.2.8",
- "bitflags",
+ "bitflags 1.3.2",
  "bytes",
  "futures-util",
  "http",
@@ -3695,7 +3716,7 @@ checksum = "08b108ad2665fa3f6e6a517c3d80ec3e77d224c47d605167aefaa5d7ef97fa48"
 dependencies = [
  "async-trait",
  "axum-core 0.3.0",
- "bitflags",
+ "bitflags 1.3.2",
  "bytes",
  "futures-util",
  "http",
@@ -3875,7 +3896,7 @@ version = "0.60.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "062dddbc1ba4aca46de6338e2bf87771414c335f7b2f2036e8f3e9befebf88e6"
 dependencies = [
- "bitflags",
+ "bitflags 1.3.2",
  "cexpr",
  "clang-sys",
  "lazy_static 1.4.0",
@@ -3909,6 +3930,12 @@ version = "1.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
 
+[[package]]
+name = "bitflags"
+version = "2.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "487f1e0fcbe47deb8b0574e646def1c903389d95241dd1bbcc6ce4a715dfc0c1"
+
 [[package]]
 name = "bitmaps"
 version = "2.1.0"
@@ -4331,7 +4358,7 @@ checksum = "a0610544180c38b88101fecf2dd634b174a62eef6946f84dfc6a7127512b381c"
 dependencies = [
  "ansi_term 0.12.1",
  "atty",
- "bitflags",
+ "bitflags 1.3.2",
  "strsim 0.8.0",
  "textwrap 0.11.0",
  "unicode-width",
@@ -4345,9 +4372,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "71655c45cb9845d3270c9d6df84ebe72b4dad3c2ba3f7023ad47c144e4e473a5"
 dependencies = [
  "atty",
- "bitflags",
- "clap_derive",
- "clap_lex",
+ "bitflags 1.3.2",
+ "clap_derive 3.2.18",
+ "clap_lex 0.2.4",
  "indexmap",
  "once_cell",
  "strsim 0.10.0",
@@ -4355,6 +4382,21 @@ dependencies = [
  "textwrap 0.16.0",
 ]
 
+[[package]]
+name = "clap"
+version = "4.1.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "42dfd32784433290c51d92c438bb72ea5063797fc3cc9a21a8c4346bebbb2098"
+dependencies = [
+ "bitflags 2.0.2",
+ "clap_derive 4.1.9",
+ "clap_lex 0.3.3",
+ "is-terminal",
+ "once_cell",
+ "strsim 0.10.0",
+ "termcolor",
+]
+
 [[package]]
 name = "clap_complete"
 version = "3.2.4"
@@ -4377,6 +4419,19 @@ dependencies = [
  "syn 1.0.105",
 ]
 
+[[package]]
+name = "clap_derive"
+version = "4.1.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fddf67631444a3a3e3e5ac51c36a5e01335302de677bd78759eaa90ab1f46644"
+dependencies = [
+ "heck 0.4.0",
+ "proc-macro-error",
+ "proc-macro2 1.0.47",
+ "quote 1.0.21",
+ "syn 1.0.105",
+]
+
 [[package]]
 name = "clap_lex"
 version = "0.2.4"
@@ -4386,6 +4441,15 @@ dependencies = [
  "os_str_bytes",
 ]
 
+[[package]]
+name = "clap_lex"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "033f6b7a4acb1f358c742aaca805c939ee73b4c6209ae4318ec7aca81c42e646"
+dependencies = [
+ "os_str_bytes",
+]
+
 [[package]]
 name = "cloud-storage"
 version = "0.11.1"
@@ -4763,7 +4827,7 @@ version = "0.21.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "486d44227f71a1ef39554c0dc47e44b9f4139927c75043312690c3f476d1d788"
 dependencies = [
- "bitflags",
+ "bitflags 1.3.2",
  "crossterm_winapi 0.8.0",
  "libc",
  "mio 0.7.14",
@@ -4779,7 +4843,7 @@ version = "0.22.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c85525306c4291d1b73ce93c8acf9c339f9b213aef6c1d85c3830cbf1c16325c"
 dependencies = [
- "bitflags",
+ "bitflags 1.3.2",
  "crossterm_winapi 0.9.0",
  "libc",
  "mio 0.7.14",
@@ -5093,7 +5157,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "01e2adfd0a7a81070ed7beec0c62636458926326c16fedb77796d41e447b282d"
 dependencies = [
  "bigdecimal",
- "bitflags",
+ "bitflags 1.3.2",
  "byteorder",
  "chrono",
  "diesel_derives",
@@ -5420,6 +5484,15 @@ dependencies = [
  "termcolor",
 ]
 
+[[package]]
+name = "env_logger"
+version = "0.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "85cdab6a89accf66733ad5a1693a4dcced6aeff64602b634530dd73c1f3ee9f0"
+dependencies = [
+ "log",
+]
+
 [[package]]
 name = "environmental"
 version = "1.1.4"
@@ -6046,7 +6119,7 @@ version = "0.15.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2994bee4a3a6a51eb90c218523be382fd7ea09b16380b9312e9dbe955ff7c7d1"
 dependencies = [
- "bitflags",
+ "bitflags 1.3.2",
  "libc",
  "libgit2-sys",
  "log",
@@ -6059,7 +6132,7 @@ version = "0.16.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ccf7f68c2995f392c49fffb4f95ae2c873297830eb25c6bc4c114ce8f4562acc"
 dependencies = [
- "bitflags",
+ "bitflags 1.3.2",
  "libc",
  "libgit2-sys",
  "log",
@@ -6093,7 +6166,7 @@ version = "0.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "93e3af942408868f6934a7b85134a3230832b9977cf66125df2f9edcfce4ddcc"
 dependencies = [
- "bitflags",
+ "bitflags 1.3.2",
  "ignore",
  "walkdir",
 ]
@@ -6212,7 +6285,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4cff78e5788be1e0ab65b04d306b2ed5092c815ec97ec70f4ebd5aee158aa55d"
 dependencies = [
  "base64 0.13.0",
- "bitflags",
+ "bitflags 1.3.2",
  "bytes",
  "headers-core",
  "http",
@@ -6254,6 +6327,12 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "hermit-abi"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fed44880c466736ef9a5c5b5facefb5ed0785676d0c02d612db14e54f0d84286"
+
 [[package]]
 name = "hex"
 version = "0.4.3"
@@ -6702,6 +6781,29 @@ version = "1.0.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "adab1eaa3408fb7f0c777a73e7465fd5656136fc93b670eb6df3c88c2c1344e3"
 
+[[package]]
+name = "inferno"
+version = "0.11.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2fb7c1b80a1dfa604bb4a649a5c5aeef3d913f7c520cb42b40e534e8a61bcdfc"
+dependencies = [
+ "ahash 0.8.3",
+ "clap 4.1.11",
+ "crossbeam-channel",
+ "crossbeam-utils",
+ "dashmap",
+ "env_logger 0.10.0",
+ "indexmap",
+ "is-terminal",
+ "itoa",
+ "log",
+ "num-format",
+ "once_cell",
+ "quick-xml 0.26.0",
+ "rgb",
+ "str_stack",
+]
+
 [[package]]
 name = "instant"
 version = "0.1.12"
@@ -6739,6 +6841,17 @@ version = "0.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1ea37f355c05dde75b84bba2d767906ad522e97cd9e2eef2be7a4ab7fb442c06"
 
+[[package]]
+name = "io-lifetimes"
+version = "1.0.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "09270fd4fa1111bc614ed2246c7ef56239a3063d5be0d1ec3b589c505d400aeb"
+dependencies = [
+ "hermit-abi 0.3.1",
+ "libc",
+ "windows-sys 0.45.0",
+]
+
 [[package]]
 name = "ipnet"
 version = "2.5.0"
@@ -6754,6 +6867,18 @@ dependencies = [
  "ipnet",
 ]
 
+[[package]]
+name = "is-terminal"
+version = "0.4.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8687c819457e979cc940d09cb16e42a1bf70aa6b60a549de6d3a62a0ee90c69e"
+dependencies = [
+ "hermit-abi 0.3.1",
+ "io-lifetimes 1.0.9",
+ "rustix 0.36.11",
+ "windows-sys 0.45.0",
+]
+
 [[package]]
 name = "is_debug"
 version = "1.0.1"
@@ -6798,9 +6923,9 @@ dependencies = [
 
 [[package]]
 name = "itoa"
-version = "1.0.3"
+version = "1.0.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6c8af84674fe1f223a982c933a0ee1086ac4d4052aa0fb8060c12c6ad838e754"
+checksum = "453ad9f582a441959e5f0d088b02ce04cfe8d51a8eaf077f12ac6d3e94164ca6"
 
 [[package]]
 name = "jemalloc-sys"
@@ -7049,7 +7174,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6607c62aa161d23d17a9072cc5da0be67cdfc89d3afb1e8d9c842bebc2525ffe"
 dependencies = [
  "arrayvec 0.5.2",
- "bitflags",
+ "bitflags 1.3.2",
  "cfg-if",
  "ryu",
  "static_assertions",
@@ -7057,9 +7182,9 @@ dependencies = [
 
 [[package]]
 name = "libc"
-version = "0.2.132"
+version = "0.2.140"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8371e4e5341c3a96db127eb2465ac681ced4c433e01dd0e938adbef26ba93ba5"
+checksum = "99227334921fae1a979cf0bfdfcc6b3e5ce376ef57e16fb6fb3ea2ed6095f80c"
 
 [[package]]
 name = "libfuzzer-sys"
@@ -7228,6 +7353,12 @@ version = "0.0.46"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d4d2456c373231a208ad294c33dc5bff30051eafd954cd4caae83a712b12854d"
 
+[[package]]
+name = "linux-raw-sys"
+version = "0.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f051f77a7c8e6957c0696eac88f26b0117e54f52d3fc682ab19397a8812846a4"
+
 [[package]]
 name = "listener"
 version = "0.1.0"
@@ -7306,7 +7437,7 @@ version = "0.90.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6f3734ab1d7d157fc0c45110e06b587c31cd82bea2ccfd6b563cbff0aaeeb1d3"
 dependencies = [
- "bitflags",
+ "bitflags 1.3.2",
  "serde 1.0.149",
  "serde_json",
  "serde_repr",
@@ -7489,7 +7620,7 @@ dependencies = [
  "libc",
  "log",
  "wasi 0.11.0+wasi-snapshot-preview1",
- "windows-sys",
+ "windows-sys 0.36.1",
 ]
 
 [[package]]
@@ -8674,6 +8805,16 @@ dependencies = [
  "syn 1.0.105",
 ]
 
+[[package]]
+name = "num-format"
+version = "0.4.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a652d9771a63711fd3c3deb670acfbe5c30a4072e664d7a3bf5a9e1056ac72c3"
+dependencies = [
+ "arrayvec 0.7.2",
+ "itoa",
+]
+
 [[package]]
 name = "num-integer"
 version = "0.1.45"
@@ -8731,7 +8872,7 @@ version = "1.13.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "19e64526ebdee182341572e50e9ad03965aa510cd94427a4549448f285e957a1"
 dependencies = [
- "hermit-abi",
+ "hermit-abi 0.1.19",
  "libc",
 ]
 
@@ -8795,7 +8936,7 @@ version = "0.10.41"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "618febf65336490dfcf20b73f885f5651a0c89c64c2d4a8c3662585a70bf5bd0"
 dependencies = [
- "bitflags",
+ "bitflags 1.3.2",
  "cfg-if",
  "foreign-types",
  "libc",
@@ -8990,7 +9131,7 @@ dependencies = [
  "libc",
  "redox_syscall 0.2.16",
  "smallvec",
- "windows-sys",
+ "windows-sys 0.36.1",
 ]
 
 [[package]]
@@ -9249,7 +9390,7 @@ version = "0.17.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5d708eaf860a19b19ce538740d2b4bdeeb8337fa53f7738455e706623ad5c638"
 dependencies = [
- "bitflags",
+ "bitflags 1.3.2",
  "crc32fast",
  "flate2",
  "miniz_oxide 0.6.2",
@@ -9569,13 +9710,13 @@ version = "0.14.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2dfb6451c91904606a1abe93e83a8ec851f45827fa84273f256ade45dc095818"
 dependencies = [
- "bitflags",
+ "bitflags 1.3.2",
  "byteorder",
  "chrono",
  "flate2",
  "hex",
  "lazy_static 1.4.0",
- "rustix",
+ "rustix 0.35.9",
 ]
 
 [[package]]
@@ -9629,7 +9770,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1e0d9cc07f18492d879586c92b485def06bc850da3118075cd45d50e9c95b0e5"
 dependencies = [
  "bit-set",
- "bitflags",
+ "bitflags 1.3.2",
  "byteorder",
  "lazy_static 1.4.0",
  "num-traits 0.2.15",
@@ -9764,6 +9905,15 @@ dependencies = [
  "memchr",
 ]
 
+[[package]]
+name = "quick-xml"
+version = "0.26.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7f50b1c63b38611e7d4d7f68b82d3ad0cc71a2ad2e7f61fc10f1328d917c93cd"
+dependencies = [
+ "memchr",
+]
+
 [[package]]
 name = "quote"
 version = "0.6.13"
@@ -9916,7 +10066,7 @@ version = "10.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6aa2540135b6a94f74c7bc90ad4b794f822026a894f3d7bcd185c100d13d4ad6"
 dependencies = [
- "bitflags",
+ "bitflags 1.3.2",
 ]
 
 [[package]]
@@ -10012,7 +10162,7 @@ version = "0.2.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "fb5a58c1855b4b6819d59012155603f0b22ad30cad752600aadfcb695265519a"
 dependencies = [
- "bitflags",
+ "bitflags 1.3.2",
 ]
 
 [[package]]
@@ -10339,12 +10489,26 @@ version = "0.35.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "72c825b8aa8010eb9ee99b75f05e10180b9278d161583034d7574c9d617aeada"
 dependencies = [
- "bitflags",
+ "bitflags 1.3.2",
+ "errno",
+ "io-lifetimes 0.7.3",
+ "libc",
+ "linux-raw-sys 0.0.46",
+ "windows-sys 0.36.1",
+]
+
+[[package]]
+name = "rustix"
+version = "0.36.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "db4165c9963ab29e422d6c26fbc1d37f15bace6b2810221f9d925023480fcf0e"
+dependencies = [
+ "bitflags 1.3.2",
  "errno",
- "io-lifetimes",
+ "io-lifetimes 1.0.9",
  "libc",
- "linux-raw-sys",
- "windows-sys",
+ "linux-raw-sys 0.1.4",
+ "windows-sys 0.45.0",
 ]
 
 [[package]]
@@ -10469,7 +10633,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "88d6731146462ea25d9244b2ed5fd1d716d25c52e4d54aa4fb0f3c4e9854dbe2"
 dependencies = [
  "lazy_static 1.4.0",
- "windows-sys",
+ "windows-sys 0.36.1",
 ]
 
 [[package]]
@@ -10515,7 +10679,7 @@ version = "2.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2bc1bb97804af6631813c55739f771071e0f2ed33ee20b68c86ec505d906356c"
 dependencies = [
- "bitflags",
+ "bitflags 1.3.2",
  "core-foundation",
  "core-foundation-sys",
  "libc",
@@ -10541,7 +10705,7 @@ dependencies = [
  "hyper",
  "indicatif 0.17.3",
  "log",
- "quick-xml",
+ "quick-xml 0.22.0",
  "regex",
  "reqwest",
  "semver",
@@ -11125,6 +11289,12 @@ dependencies = [
  "atty",
 ]
 
+[[package]]
+name = "str_stack"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9091b6114800a5f2141aee1d1b9d6ca3592ac062dc5decb3764ec5895a47b4eb"
+
 [[package]]
 name = "string_cache"
 version = "0.8.4"
@@ -11798,7 +11968,7 @@ version = "0.3.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3c530c8675c1dbf98facee631536fa116b5fb6382d7dd6dc1b118d970eafe3ba"
 dependencies = [
- "bitflags",
+ "bitflags 1.3.2",
  "bytes",
  "futures-core",
  "futures-util",
@@ -11942,7 +12112,7 @@ version = "0.17.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "23ed0a32c88b039b73f1b6c5acbd0554bfa5b6be94467375fd947c4de3a02271"
 dependencies = [
- "bitflags",
+ "bitflags 1.3.2",
  "cassowary",
  "crossterm 0.22.1",
  "unicode-segmentation",
@@ -12213,7 +12383,7 @@ dependencies = [
  "errno",
  "js-sys",
  "libc",
- "rustix",
+ "rustix 0.35.9",
  "wasi 0.11.0+wasi-snapshot-preview1",
  "wasm-bindgen",
  "winapi 0.3.9",
@@ -12568,43 +12738,109 @@ version = "0.36.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ea04155a16a59f9eab786fe12a4a450e75cdb175f9e0d80da1e17db09f55b8d2"
 dependencies = [
- "windows_aarch64_msvc",
- "windows_i686_gnu",
- "windows_i686_msvc",
- "windows_x86_64_gnu",
- "windows_x86_64_msvc",
+ "windows_aarch64_msvc 0.36.1",
+ "windows_i686_gnu 0.36.1",
+ "windows_i686_msvc 0.36.1",
+ "windows_x86_64_gnu 0.36.1",
+ "windows_x86_64_msvc 0.36.1",
+]
+
+[[package]]
+name = "windows-sys"
+version = "0.45.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "75283be5efb2831d37ea142365f009c02ec203cd29a3ebecbc093d52315b66d0"
+dependencies = [
+ "windows-targets",
+]
+
+[[package]]
+name = "windows-targets"
+version = "0.42.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8e5180c00cd44c9b1c88adb3693291f1cd93605ded80c250a75d472756b4d071"
+dependencies = [
+ "windows_aarch64_gnullvm",
+ "windows_aarch64_msvc 0.42.2",
+ "windows_i686_gnu 0.42.2",
+ "windows_i686_msvc 0.42.2",
+ "windows_x86_64_gnu 0.42.2",
+ "windows_x86_64_gnullvm",
+ "windows_x86_64_msvc 0.42.2",
 ]
 
+[[package]]
+name = "windows_aarch64_gnullvm"
+version = "0.42.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "597a5118570b68bc08d8d59125332c54f1ba9d9adeedeef5b99b02ba2b0698f8"
+
 [[package]]
 name = "windows_aarch64_msvc"
 version = "0.36.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9bb8c3fd39ade2d67e9874ac4f3db21f0d710bee00fe7cab16949ec184eeaa47"
 
+[[package]]
+name = "windows_aarch64_msvc"
+version = "0.42.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e08e8864a60f06ef0d0ff4ba04124db8b0fb3be5776a5cd47641e942e58c4d43"
+
 [[package]]
 name = "windows_i686_gnu"
 version = "0.36.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "180e6ccf01daf4c426b846dfc66db1fc518f074baa793aa7d9b9aaeffad6a3b6"
 
+[[package]]
+name = "windows_i686_gnu"
+version = "0.42.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c61d927d8da41da96a81f029489353e68739737d3beca43145c8afec9a31a84f"
+
 [[package]]
 name = "windows_i686_msvc"
 version = "0.36.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e2e7917148b2812d1eeafaeb22a97e4813dfa60a3f8f78ebe204bcc88f12f024"
 
+[[package]]
+name = "windows_i686_msvc"
+version = "0.42.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "44d840b6ec649f480a41c8d80f9c65108b92d89345dd94027bfe06ac444d1060"
+
 [[package]]
 name = "windows_x86_64_gnu"
 version = "0.36.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4dcd171b8776c41b97521e5da127a2d86ad280114807d0b2ab1e462bc764d9e1"
 
+[[package]]
+name = "windows_x86_64_gnu"
+version = "0.42.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8de912b8b8feb55c064867cf047dda097f92d51efad5b491dfb98f6bbb70cb36"
+
+[[package]]
+name = "windows_x86_64_gnullvm"
+version = "0.42.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "26d41b46a36d453748aedef1486d5c7a85db22e56aff34643984ea85514e94a3"
+
 [[package]]
 name = "windows_x86_64_msvc"
 version = "0.36.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c811ca4a8c853ef420abd8592ba53ddbbac90410fab6903b3e79972a631f7680"
 
+[[package]]
+name = "windows_x86_64_msvc"
+version = "0.42.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9aec5da331524158c6d1a4ac0ab1541149c0b9505fde06423b02f5ef0106b9f0"
+
 [[package]]
 name = "winreg"
 version = "0.10.1"
diff --git a/Cargo.toml b/Cargo.toml
index 56c92ed75a5c7..8dc827daadb5e 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -9,6 +9,7 @@ members = [
     "aptos-move/aptos-aggregator",
     "aptos-move/aptos-debugger",
     "aptos-move/aptos-gas",
+    "aptos-move/aptos-gas-profiling",
     "aptos-move/aptos-release-builder",
     "aptos-move/aptos-resource-viewer",
     "aptos-move/aptos-sdk-builder",
@@ -282,6 +283,7 @@ aptos-framework = { path = "aptos-move/framework" }
 aptos-fuzzer = { path = "testsuite/aptos-fuzzer" }
 aptos-gas = { path = "aptos-move/aptos-gas" }
 aptos-gas-algebra-ext = { path = "aptos-move/gas-algebra-ext" }
+aptos-gas-profiling = { path = "aptos-move/aptos-gas-profiling" }
 aptos-genesis = { path = "crates/aptos-genesis" }
 aptos-github-client = { path = "secure/storage/github" }
 aptos-global-constants = { path = "config/global-constants" }
@@ -454,6 +456,7 @@ hyper-tls = "0.5.0"
 include_dir = { version = "0.7.2", features = ["glob"] }
 indicatif = "0.15.0"
 indoc = "1.0.6"
+inferno = "0.11.14"
 ipnet = "2.5.0"
 itertools = "0.10.3"
 jemallocator = { version = "0.3.2", features = [
diff --git a/aptos-move/aptos-debugger/Cargo.toml b/aptos-move/aptos-debugger/Cargo.toml
index e73051012fe0f..73b83abc0d0c6 100644
--- a/aptos-move/aptos-debugger/Cargo.toml
+++ b/aptos-move/aptos-debugger/Cargo.toml
@@ -13,6 +13,7 @@ edition = "2021"
 anyhow = { workspace = true }
 aptos-crypto = { workspace = true }
 aptos-gas = { workspace = true }
+aptos-gas-profiling = { workspace = true }
 aptos-logger = { workspace = true }
 aptos-resource-viewer = { workspace = true }
 aptos-rest-client = { workspace = true }
@@ -20,6 +21,7 @@ aptos-state-view = { workspace = true }
 aptos-types = { workspace = true }
 aptos-validator-interface = { workspace = true }
 aptos-vm = { workspace = true }
+aptos-vm-logging = { workspace = true }
 clap = { workspace = true }
 move-binary-format = { workspace = true }
 move-cli = { workspace = true }
diff --git a/aptos-move/aptos-debugger/src/lib.rs b/aptos-move/aptos-debugger/src/lib.rs
index 0844bd7927019..2dc40c7c7d1be 100644
--- a/aptos-move/aptos-debugger/src/lib.rs
+++ b/aptos-move/aptos-debugger/src/lib.rs
@@ -3,16 +3,22 @@
 
 use anyhow::{format_err, Result};
 use aptos_gas::{
-    AbstractValueSizeGasParameters, ChangeSetConfigs, NativeGasParameters,
+    AbstractValueSizeGasParameters, ChangeSetConfigs, NativeGasParameters, StandardGasMeter,
     LATEST_GAS_FEATURE_VERSION,
 };
+use aptos_gas_profiling::{GasProfiler, TransactionGasLog};
 use aptos_resource_viewer::{AnnotatedAccountStateBlob, AptosValueAnnotator};
 use aptos_rest_client::Client;
+use aptos_state_view::TStateView;
 use aptos_types::{
     account_address::AccountAddress,
     chain_id::ChainId,
     on_chain_config::{Features, OnChainConfig, TimedFeatures},
-    transaction::{ChangeSet, Transaction, TransactionInfo, TransactionOutput, Version},
+    transaction::{
+        ChangeSet, SignedTransaction, Transaction, TransactionInfo, TransactionOutput,
+        TransactionPayload, Version,
+    },
+    vm_status::VMStatus,
 };
 use aptos_validator_interface::{
     AptosValidatorInterface, DBDebuggerInterface, DebuggerStateView, RestDebuggerInterface,
@@ -22,6 +28,7 @@ use aptos_vm::{
     move_vm_ext::{MoveVmExt, SessionExt, SessionId},
     AptosVM, VMExecutor,
 };
+use aptos_vm_logging::log_schema::AdapterLogSchema;
 use move_binary_format::errors::VMResult;
 use std::{path::Path, sync::Arc};
 
@@ -54,6 +61,47 @@ impl AptosDebugger {
             .map_err(|err| format_err!("Unexpected VM Error: {:?}", err))
     }
 
+    pub fn execute_transaction_at_version_with_gas_profiler(
+        &self,
+        version: Version,
+        txn: SignedTransaction,
+    ) -> Result<(VMStatus, TransactionOutput, TransactionGasLog)> {
+        let state_view = DebuggerStateView::new(self.debugger.clone(), version);
+        let log_context = AdapterLogSchema::new(state_view.id(), 0);
+        let txn = txn
+            .check_signature()
+            .map_err(|err| format_err!("Unexpected VM Error: {:?}", err))?;
+
+        let (status, output, gas_profiler) =
+            AptosVM::execute_user_transaction_with_custom_gas_meter(
+                &state_view,
+                &txn,
+                &log_context,
+                |gas_feature_version, gas_params, storage_gas_params, balance| {
+                    let gas_meter = StandardGasMeter::new(
+                        gas_feature_version,
+                        gas_params,
+                        storage_gas_params,
+                        balance,
+                    );
+                    let gas_profiler = match txn.payload() {
+                        TransactionPayload::Script(_) => GasProfiler::new_script(gas_meter),
+                        TransactionPayload::EntryFunction(entry_func) => GasProfiler::new_function(
+                            gas_meter,
+                            entry_func.module().clone(),
+                            entry_func.function().to_owned(),
+                            entry_func.ty_args().to_vec(),
+                        ),
+                        TransactionPayload::ModuleBundle(..) => unreachable!("not supported"),
+                        TransactionPayload::Multisig(..) => unimplemented!("not supported yet"),
+                    };
+                    Ok(gas_profiler)
+                },
+            )?;
+
+        Ok((status, output, gas_profiler.finish()))
+    }
+
     pub async fn execute_past_transactions(
         &self,
         mut begin: Version,
diff --git a/aptos-move/aptos-gas-profiling/Cargo.toml b/aptos-move/aptos-gas-profiling/Cargo.toml
new file mode 100644
index 0000000000000..1f0414bbf46a2
--- /dev/null
+++ b/aptos-move/aptos-gas-profiling/Cargo.toml
@@ -0,0 +1,26 @@
+[package]
+name = "aptos-gas-profiling"
+version = "0.1.0"
+
+# Workspace inherited keys
+authors = { workspace = true }
+edition = { workspace = true }
+homepage = { workspace = true }
+license = { workspace = true }
+publish = { workspace = true }
+repository = { workspace = true }
+rust-version = { workspace = true }
+
+[dependencies]
+anyhow = { workspace = true }
+inferno = { workspace = true }
+regex = { workspace = true }
+
+aptos-framework = { workspace = true }
+aptos-gas = { workspace = true }
+aptos-package-builder = { workspace = true }
+aptos-types = { workspace = true }
+
+move-binary-format = { workspace = true }
+move-core-types = { workspace = true }
+move-vm-types = { workspace = true }
diff --git a/aptos-move/aptos-gas-profiling/README.md b/aptos-move/aptos-gas-profiling/README.md
new file mode 100644
index 0000000000000..cbd4357876c4e
--- /dev/null
+++ b/aptos-move/aptos-gas-profiling/README.md
@@ -0,0 +1,41 @@
+# Gas Profiling
+
+## Overview
+This crate implements a gas profiler that can be plugged into the Aptos VM to generate comprehensive traces of gas usage, referred to as the transaction gas log.
+It also contains a module for visualizing the transaction gas log, in the form of a flamegraph.
+
+## Running the Gas Profiler
+You can run the gas profiler by appending the `--profile-gas` option to the aptos cli's `move publish`, `move run` & `move run-script` commands. Here is an example:
+```
+>> cargo run -p aptos -- move publish --profile-gas
+    Finished dev [unoptimized + debuginfo] target(s) in 0.51s
+     Running `/home/vgao/aptos-core/target/debug/aptos move publish --profile-gas`
+Compiling, may take a little while to download git dependencies...
+BUILDING empty_fun
+package size 427 bytes
+
+Simulating transaction locally with the gas profiler...
+This is still experimental so results may be inaccurate.
+
+Execution & IO Gas flamegraph saved to gas-profiling/txn-69e19ee4-0x1-code-publish_package_txn.exec_io.svg
+Storage fee flamegraph saved to gas-profiling/txn-69e19ee4-0x1-code-publish_package_txn.storage.svg
+
+{
+  "Result": {
+    "transaction_hash": "0x69e19ee4cc89cb1f84ee21a46e6b281bd8696115aa332275eca38c4857818dfe",
+    "gas_used": 1007,
+    "gas_unit_price": 100,
+    "sender": "dbcbe741d003a7369d87ec8717afb5df425977106497052f96f4e236372f7dd5",
+    "success": true,
+    "version": 473269362,
+    "vm_status": "status EXECUTED of type Execution"
+  }
+}
+```
+
+## Performance Implications
+It is important to note that the current gas profiler implementation is quite heavy-weight since it records every Move bytecode instruction and its cost. If real-time gas profiling is required, it is recommended to develop a custom profiler that operates on aggregated data. A standard light-weight implementation may be provided in the future.
+
+## Known Issues & Future Plans
+1. While addresses are truncated in the flamegraphs, they are still somewhat cumbersome. We plan to come up with a smart rendering algorithm that omits addresses, provided that the functions/items can still be unambiguously identified.
+2. At present, the storage fee graph does not display the free quota for events. We plan to address this in a future update.
diff --git a/aptos-move/aptos-gas-profiling/src/flamegraph.rs b/aptos-move/aptos-gas-profiling/src/flamegraph.rs
new file mode 100644
index 0000000000000..cd8a583288fbf
--- /dev/null
+++ b/aptos-move/aptos-gas-profiling/src/flamegraph.rs
@@ -0,0 +1,349 @@
+// Copyright © Aptos Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::log::{
+    CallFrame, ExecutionGasEvent, FrameName, StorageFees, TransactionGasLog, WriteOpType,
+};
+use aptos_types::{
+    access_path::Path,
+    state_store::{
+        state_key::{StateKey, StateKeyInner},
+        table::TableHandle,
+    },
+};
+use inferno::flamegraph::TextTruncateDirection;
+use move_core_types::{
+    account_address::AccountAddress,
+    gas_algebra::InternalGas,
+    identifier::IdentStr,
+    language_storage::{ModuleId, TypeTag},
+};
+use regex::Captures;
+use std::{
+    fmt::{self, Display},
+    ops::Deref,
+};
+
+/// Wrapper to help render the underlying data in formats desirable by the flamegraph.
+struct Render<'a, T>(&'a T);
+
+impl<'a> Display for Render<'a, AccountAddress> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let addr_short = self.0.short_str_lossless();
+        write!(f, "0x")?;
+        if addr_short.len() > 4 {
+            write!(f, "{}..", &addr_short[..4])
+        } else {
+            write!(f, "{}", addr_short)
+        }
+    }
+}
+
+impl<'a> Display for Render<'a, ModuleId> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}::{}", Render(self.0.address()), self.0.name())
+    }
+}
+
+impl<'a> Display for Render<'a, (&'a ModuleId, &'a IdentStr, &'a [TypeTag])> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}::{}", Render(self.0 .0), self.0 .1)?;
+        if !self.0 .2.is_empty() {
+            write!(
+                f,
+                "<{}>",
+                self.0
+                     .2
+                    .iter()
+                    .map(|ty| format!("{}", ty))
+                    .collect::<Vec<_>>()
+                    .join(",")
+            )?;
+        }
+        Ok(())
+    }
+}
+
+impl Display for FrameName {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Script => write!(f, "<script>"),
+            Self::Function {
+                module_id,
+                name: fn_name,
+                ty_args,
+            } => write!(
+                f,
+                "{}",
+                Render(&(module_id, fn_name.as_ident_str(), ty_args.as_slice())),
+            ),
+        }
+    }
+}
+
+impl<'a> Display for Render<'a, Path> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self.0 {
+            Path::Code(module_id) => write!(f, "{}", Render(module_id)),
+            Path::Resource(struct_ty) => write!(f, "{}", struct_ty),
+            Path::ResourceGroup(struct_ty) => write!(f, "{}", struct_ty),
+        }
+    }
+}
+
+impl<'a> Display for Render<'a, TableHandle> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", Render(&self.0 .0))
+    }
+}
+
+struct TableKey<'a> {
+    bytes: &'a [u8],
+}
+
+impl<'a> Display for TableKey<'a> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        assert!(self.bytes.len() > 2);
+        write!(f, "0x{:02x}{:02x}..", self.bytes[0], self.bytes[1])
+    }
+}
+
+impl<'a> Display for Render<'a, StateKey> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        use StateKeyInner::*;
+
+        match self.0.deref() {
+            AccessPath(ap) => {
+                write!(f, "{}::{}", Render(&ap.address), Render(&ap.get_path()))
+            },
+            TableItem { handle, key } => {
+                write!(f, "table_item<{},{}>", Render(handle), TableKey {
+                    bytes: key
+                },)
+            },
+            Raw(..) => panic!("not supported"),
+        }
+    }
+}
+
+impl<'a> Display for Render<'a, WriteOpType> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        use WriteOpType::*;
+
+        write!(f, "{}", match self.0 {
+            Creation => "create",
+            Modification => "modify",
+            Deletion => "delete",
+        })
+    }
+}
+
+fn strip_trailing_zeros_and_decimal_point(mut s: &str) -> &str {
+    while let Some(stripped) = s.strip_suffix('0') {
+        s = stripped
+    }
+    match s.strip_suffix('.') {
+        Some(stripped) => stripped,
+        None => s,
+    }
+}
+
+struct LineBuffer(Vec<String>);
+
+impl LineBuffer {
+    fn new() -> Self {
+        Self(vec![])
+    }
+
+    fn push(&mut self, item: impl AsRef<str>, count: impl Into<u64>) {
+        let count: u64 = count.into();
+
+        if count > 0 {
+            self.0.push(format!("{} {}", item.as_ref(), count));
+        }
+    }
+
+    fn into_inner(self) -> Vec<String> {
+        self.0
+    }
+}
+
+impl StorageFees {
+    /// Convert the storage fee log into folded stack lines, which can
+    /// then be used to generate a flamegraph.
+    fn to_folded_stack_lines(&self) -> Vec<String> {
+        let mut lines = LineBuffer::new();
+
+        lines.push("transaction", self.txn_storage);
+
+        for item in &self.write_set_storage {
+            lines.push(
+                format!("write_set;{}<{}>", Render(&item.op_type), Render(&item.key)),
+                item.cost,
+            )
+        }
+
+        for event in &self.events {
+            lines.push(format!("events;{}", event.ty), event.cost)
+        }
+
+        lines.into_inner()
+    }
+
+    /// Tries to generate a flamegraph from the execution log.
+    /// None will be returned if the log is empty.
+    pub fn to_flamegraph(&self, title: String) -> anyhow::Result<Option<Vec<u8>>> {
+        let lines = self.to_folded_stack_lines();
+
+        if lines.is_empty() {
+            return Ok(None);
+        }
+
+        let mut options = inferno::flamegraph::Options::default();
+        options.flame_chart = true;
+        options.text_truncate_direction = TextTruncateDirection::Right;
+        options.color_diffusion = true;
+        options.title = title;
+
+        let mut graph_content = vec![];
+        inferno::flamegraph::from_lines(
+            &mut options,
+            lines.iter().rev().map(|s| s.as_str()),
+            &mut graph_content,
+        )?;
+        let graph_content = String::from_utf8_lossy(&graph_content);
+
+        // Inferno does not allow us to customize some of the text in the resulting graph,
+        // so we have to do it through regex replacement.
+        let re = regex::Regex::new("([1-9][0-9]*(,[0-9]+)*) samples")
+            .expect("should be able to build regex successfully");
+        let graph_content = re.replace_all(&graph_content, |caps: &Captures| {
+            let count: u64 = caps[1]
+                .replace(',', "")
+                .parse()
+                .expect("should be able parse count as u64");
+
+            format!("{} Octa", count)
+        });
+
+        Ok(Some(graph_content.as_bytes().to_vec()))
+    }
+}
+
+impl TransactionGasLog {
+    /// Convert the execution gas log into folded stack lines, which can
+    /// then be used to generate a flamegraph.
+    fn to_folded_stack_lines(&self) -> Vec<String> {
+        let mut lines = LineBuffer::new();
+
+        lines.push("intrinsic", self.intrinsic_cost);
+
+        let mut path = vec![];
+
+        struct Rec<'a> {
+            lines: &'a mut LineBuffer,
+            path: &'a mut Vec<String>,
+        }
+
+        impl<'a> Rec<'a> {
+            fn visit(&mut self, frame: &CallFrame) {
+                self.path.push(format!("{}", frame.name));
+
+                let mut frame_cost = InternalGas::new(0);
+
+                for event in &frame.events {
+                    use ExecutionGasEvent::*;
+
+                    match event {
+                        Loc(_) => (),
+                        Bytecode { cost, .. } => frame_cost += *cost,
+                        Call(inner_frame) => self.visit(inner_frame),
+                        CallNative {
+                            module_id: module,
+                            fn_name,
+                            ty_args,
+                            cost,
+                        } => self.lines.push(
+                            format!(
+                                "{};{}",
+                                self.path(),
+                                Render(&(module, fn_name.as_ident_str(), ty_args.as_slice())),
+                            ),
+                            *cost,
+                        ),
+                        LoadResource { addr, ty, cost } => self.lines.push(
+                            format!("{};load<{}::{}>", self.path(), Render(addr), ty),
+                            *cost,
+                        ),
+                    }
+                }
+
+                self.lines.push(&self.path(), frame_cost);
+                self.path.pop();
+            }
+
+            fn path(&self) -> String {
+                self.path.join(";")
+            }
+        }
+
+        Rec {
+            lines: &mut lines,
+            path: &mut path,
+        }
+        .visit(&self.call_graph);
+
+        for item in &self.write_set_transient {
+            lines.push(
+                format!("write_set;{}<{}>", Render(&item.op_type), Render(&item.key)),
+                item.cost,
+            )
+        }
+
+        lines.into_inner()
+    }
+
+    /// Tries to generate a flamegraph from the execution log.
+    /// None will be returned if the log is empty.
+    pub fn to_flamegraph(&self, title: String) -> anyhow::Result<Option<Vec<u8>>> {
+        let lines = self.to_folded_stack_lines();
+
+        if lines.is_empty() {
+            return Ok(None);
+        }
+
+        let mut options = inferno::flamegraph::Options::default();
+        options.flame_chart = true;
+        options.text_truncate_direction = TextTruncateDirection::Right;
+        options.color_diffusion = true;
+        options.title = title;
+
+        let mut graph_content = vec![];
+        inferno::flamegraph::from_lines(
+            &mut options,
+            lines.iter().rev().map(|s| s.as_str()),
+            &mut graph_content,
+        )?;
+        let graph_content = String::from_utf8_lossy(&graph_content);
+
+        // Inferno does not allow us to customize some of the text in the resulting graph,
+        // so we have to do it through regex replacement.
+        let re = regex::Regex::new("([1-9][0-9]*(,[0-9]+)*) samples")
+            .expect("should be able to build regex successfully");
+        let graph_content = re.replace_all(&graph_content, |caps: &Captures| {
+            let count: u64 = caps[1]
+                .replace(',', "")
+                .parse()
+                .expect("should be able parse count as u64");
+
+            let count_scaled = count as f64 / u64::from(self.gas_scaling_factor) as f64;
+
+            format!(
+                "{} gas units",
+                strip_trailing_zeros_and_decimal_point(&format!("{:.8}", count_scaled))
+            )
+        });
+
+        Ok(Some(graph_content.as_bytes().to_vec()))
+    }
+}
diff --git a/aptos-move/aptos-gas-profiling/src/lib.rs b/aptos-move/aptos-gas-profiling/src/lib.rs
new file mode 100644
index 0000000000000..27fa59e617d88
--- /dev/null
+++ b/aptos-move/aptos-gas-profiling/src/lib.rs
@@ -0,0 +1,9 @@
+// Copyright © Aptos Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+mod flamegraph;
+mod log;
+mod profiler;
+
+pub use log::{FrameName, TransactionGasLog};
+pub use profiler::GasProfiler;
diff --git a/aptos-move/aptos-gas-profiling/src/log.rs b/aptos-move/aptos-gas-profiling/src/log.rs
new file mode 100644
index 0000000000000..2ee1a76b734f4
--- /dev/null
+++ b/aptos-move/aptos-gas-profiling/src/log.rs
@@ -0,0 +1,139 @@
+// Copyright © Aptos Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+use aptos_gas::{Fee, GasScalingFactor};
+use aptos_types::state_store::state_key::StateKey;
+use move_binary_format::{file_format::CodeOffset, file_format_common::Opcodes};
+use move_core_types::{
+    account_address::AccountAddress,
+    gas_algebra::InternalGas,
+    identifier::Identifier,
+    language_storage::{ModuleId, TypeTag},
+};
+
+/// An event occurred during the execution of a function, along with the
+/// gas cost associated with it, if any.
+#[derive(Debug)]
+pub enum ExecutionGasEvent {
+    /// A special event indicating that the program counter has moved to
+    /// a specific offset. This is emitted by the branch instructions
+    /// and is crucial for reconstructing the control flow.
+    Loc(CodeOffset),
+    Bytecode {
+        op: Opcodes,
+        cost: InternalGas,
+    },
+    Call(CallFrame),
+    CallNative {
+        module_id: ModuleId,
+        fn_name: Identifier,
+        ty_args: Vec<TypeTag>,
+        cost: InternalGas,
+    },
+    LoadResource {
+        addr: AccountAddress,
+        ty: TypeTag,
+        cost: InternalGas,
+    },
+}
+
+/// An enum representing the name of a call frame.
+/// Could be either a script or a function.
+#[derive(Debug)]
+pub enum FrameName {
+    Script,
+    Function {
+        module_id: ModuleId,
+        name: Identifier,
+        ty_args: Vec<TypeTag>,
+    },
+}
+
+/// A struct containing information about a function call, including the name of the
+/// function and all gas events that happened during the call.
+#[derive(Debug)]
+pub struct CallFrame {
+    pub name: FrameName,
+    pub events: Vec<ExecutionGasEvent>,
+}
+
+/// The type of an operation performed on a storage item.
+///
+/// Possible values: Creation, Modification & Deletion.
+#[derive(Debug)]
+pub enum WriteOpType {
+    Creation,
+    Modification,
+    Deletion,
+}
+
+/// Struct representing the transient (IO) cost of a write operation.
+#[derive(Debug)]
+pub struct WriteTransient {
+    pub key: StateKey,
+    pub op_type: WriteOpType,
+    pub cost: InternalGas,
+}
+
+/// Struct representing the storage cost of a write operation.
+#[derive(Debug)]
+pub struct WriteStorage {
+    pub key: StateKey,
+    pub op_type: WriteOpType,
+    pub cost: Fee,
+}
+
+#[derive(Debug)]
+/// Struct representing the storage cost of an event.
+pub struct EventStorage {
+    pub ty: TypeTag,
+    pub cost: Fee,
+}
+
+#[derive(Debug)]
+// Struct containing all types of storage fees.
+pub struct StorageFees {
+    pub write_set_storage: Vec<WriteStorage>,
+    pub events: Vec<EventStorage>,
+    pub event_discount: Fee,
+    pub txn_storage: Fee,
+}
+
+/// A complete log that contains all gas-related information about a transaction, including
+/// the intrinsic cost, a detailed execution log and the write set costs.
+#[derive(Debug)]
+pub struct TransactionGasLog {
+    // TODO(Gas): move the execution-related entries into a separate struct.
+    pub gas_scaling_factor: GasScalingFactor,
+    pub intrinsic_cost: InternalGas,
+    pub call_graph: CallFrame,
+    pub write_set_transient: Vec<WriteTransient>,
+
+    pub storage: StorageFees,
+}
+
+impl CallFrame {
+    pub fn new_function(module_id: ModuleId, name: Identifier, ty_args: Vec<TypeTag>) -> Self {
+        Self {
+            name: FrameName::Function {
+                module_id,
+                name,
+                ty_args,
+            },
+            events: vec![],
+        }
+    }
+
+    pub fn new_script() -> Self {
+        Self {
+            name: FrameName::Script,
+            events: vec![],
+        }
+    }
+}
+
+impl TransactionGasLog {
+    pub fn entry_point(&self) -> &FrameName {
+        &self.call_graph.name
+    }
+}
diff --git a/aptos-move/aptos-gas-profiling/src/profiler.rs b/aptos-move/aptos-gas-profiling/src/profiler.rs
new file mode 100644
index 0000000000000..c682cd882ddb6
--- /dev/null
+++ b/aptos-move/aptos-gas-profiling/src/profiler.rs
@@ -0,0 +1,608 @@
+// Copyright © Aptos Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::log::{
+    CallFrame, EventStorage, ExecutionGasEvent, FrameName, StorageFees, TransactionGasLog,
+    WriteOpType, WriteStorage, WriteTransient,
+};
+use aptos_gas::{AptosGasMeter, Fee, Gas, GasScalingFactor};
+use aptos_types::{
+    contract_event::ContractEvent, state_store::state_key::StateKey, write_set::WriteOp,
+};
+use move_binary_format::{
+    errors::{Location, PartialVMResult, VMResult},
+    file_format::CodeOffset,
+    file_format_common::Opcodes,
+};
+use move_core_types::{
+    account_address::AccountAddress,
+    gas_algebra::{InternalGas, NumArgs, NumBytes},
+    identifier::Identifier,
+    language_storage::{ModuleId, TypeTag},
+};
+use move_vm_types::{
+    gas::{GasMeter, SimpleInstruction},
+    views::{TypeView, ValueView},
+};
+
+/// A special gas meter adapter that records all gas-related events, along with the associated costs
+/// assessed by the underlying gas meter.
+pub struct GasProfiler<G> {
+    base: G,
+
+    intrinsic_cost: Option<InternalGas>,
+    frames: Vec<CallFrame>,
+    write_set_transient: Vec<WriteTransient>,
+    storage_fees: Option<StorageFees>,
+}
+
+// TODO: consider switching to a library like https://docs.rs/delegate/latest/delegate/.
+macro_rules! delegate {
+    ($(
+        fn $fn: ident $(<$($lt: lifetime),*>)? (&self $(, $arg: ident : $ty: ty)* $(,)?) -> $ret_ty: ty;
+    )*) => {
+        $(fn $fn $(<$($lt)*>)? (&self, $($arg: $ty),*) -> $ret_ty {
+            self.base.$fn($($arg),*)
+        })*
+    };
+}
+
+macro_rules! delegate_mut {
+    ($(
+        fn $fn: ident $(<$($lt: lifetime),*>)? (&mut self $(, $arg: ident : $ty: ty)* $(,)?) -> $ret_ty: ty;
+    )*) => {
+        $(fn $fn $(<$($lt)*>)? (&mut self, $($arg: $ty),*) -> $ret_ty {
+            self.base.$fn($($arg),*)
+        })*
+    };
+}
+
+macro_rules! record_bytecode {
+    ($(
+        $([$op: expr])?
+        fn $fn: ident $(<$($lt: lifetime),*>)? (&mut self $(, $arg: ident : $ty: ty)* $(,)?) -> PartialVMResult<()>;
+    )*) => {
+        $(fn $fn $(<$($lt)*>)? (&mut self, $($arg: $ty),*) -> PartialVMResult<()> {
+            #[allow(unused)]
+            use Opcodes::*;
+
+            #[allow(unused)]
+            let (cost, res) = self.delegate_charge(|base| base.$fn($($arg),*));
+
+            $(
+                self.record_bytecode($op, cost);
+            )?
+
+            res
+        })*
+    };
+}
+
+impl<G> GasProfiler<G> {
+    pub fn new_script(base: G) -> Self {
+        Self {
+            base,
+
+            intrinsic_cost: None,
+            frames: vec![CallFrame::new_script()],
+            write_set_transient: vec![],
+            storage_fees: None,
+        }
+    }
+
+    pub fn new_function(
+        base: G,
+        module_id: ModuleId,
+        func_name: Identifier,
+        ty_args: Vec<TypeTag>,
+    ) -> Self {
+        Self {
+            base,
+
+            intrinsic_cost: None,
+            frames: vec![CallFrame::new_function(module_id, func_name, ty_args)],
+            write_set_transient: vec![],
+            storage_fees: None,
+        }
+    }
+}
+
+impl<G> GasProfiler<G>
+where
+    G: AptosGasMeter,
+{
+    fn active_event_stream(&mut self) -> &mut Vec<ExecutionGasEvent> {
+        &mut self.frames.last_mut().unwrap().events
+    }
+
+    fn record_bytecode(&mut self, op: Opcodes, cost: InternalGas) {
+        self.active_event_stream()
+            .push(ExecutionGasEvent::Bytecode { op, cost });
+    }
+
+    fn record_offset(&mut self, offset: CodeOffset) {
+        self.active_event_stream()
+            .push(ExecutionGasEvent::Loc(offset));
+    }
+
+    /// Delegate the charging call to the base gas meter and measure variation in balance.
+    fn delegate_charge<F, R>(&mut self, charge: F) -> (InternalGas, R)
+    where
+        F: FnOnce(&mut G) -> R,
+    {
+        let old = self.base.balance_internal();
+        let res = charge(&mut self.base);
+        let new = self.base.balance_internal();
+        let cost = old.checked_sub(new).expect("gas cost must be non-negative");
+
+        (cost, res)
+    }
+}
+
+impl<G> GasMeter for GasProfiler<G>
+where
+    G: AptosGasMeter,
+{
+    delegate_mut! {
+        // Note: we only use this callback for memory tracking, not for charging gas.
+        fn charge_ld_const_after_deserialization(&mut self, val: impl ValueView)
+            -> PartialVMResult<()>;
+
+        // Note: we don't use this to charge gas so no need to record anything.
+        fn charge_native_function_before_execution(
+            &mut self,
+            ty_args: impl ExactSizeIterator<Item = impl TypeView> + Clone,
+            args: impl ExactSizeIterator<Item = impl ValueView> + Clone,
+        ) -> PartialVMResult<()>;
+
+        // Note: we don't use this to charge gas so no need to record anything.
+        fn charge_drop_frame(
+            &mut self,
+            locals: impl Iterator<Item = impl ValueView> + Clone,
+        ) -> PartialVMResult<()>;
+    }
+
+    record_bytecode! {
+        [POP]
+        fn charge_pop(&mut self, popped_val: impl ValueView) -> PartialVMResult<()>;
+
+        [LD_CONST]
+        fn charge_ld_const(&mut self, size: NumBytes) -> PartialVMResult<()>;
+
+        [COPY_LOC]
+        fn charge_copy_loc(&mut self, val: impl ValueView) -> PartialVMResult<()>;
+
+        [MOVE_LOC]
+        fn charge_move_loc(&mut self, val: impl ValueView) -> PartialVMResult<()>;
+
+        [ST_LOC]
+        fn charge_store_loc(&mut self, val: impl ValueView) -> PartialVMResult<()>;
+
+        [PACK]
+        fn charge_pack(
+            &mut self,
+            is_generic: bool,
+            args: impl ExactSizeIterator<Item = impl ValueView> + Clone,
+        ) -> PartialVMResult<()>;
+
+        [UNPACK]
+        fn charge_unpack(
+            &mut self,
+            is_generic: bool,
+            args: impl ExactSizeIterator<Item = impl ValueView> + Clone,
+        ) -> PartialVMResult<()>;
+
+        [READ_REF]
+        fn charge_read_ref(&mut self, val: impl ValueView) -> PartialVMResult<()>;
+
+        [WRITE_REF]
+        fn charge_write_ref(
+            &mut self,
+            new_val: impl ValueView,
+            old_val: impl ValueView,
+        ) -> PartialVMResult<()>;
+
+        [EQ]
+        fn charge_eq(&mut self, lhs: impl ValueView, rhs: impl ValueView) -> PartialVMResult<()>;
+
+        [NEQ]
+        fn charge_neq(&mut self, lhs: impl ValueView, rhs: impl ValueView) -> PartialVMResult<()>;
+
+        [
+            match (is_mut, is_generic) {
+                (false, false) => IMM_BORROW_GLOBAL,
+                (false, true) => IMM_BORROW_GLOBAL_GENERIC,
+                (true, false) => MUT_BORROW_GLOBAL,
+                (true, true) => MUT_BORROW_GLOBAL_GENERIC
+            }
+        ]
+        fn charge_borrow_global(
+            &mut self,
+            is_mut: bool,
+            is_generic: bool,
+            ty: impl TypeView,
+            is_success: bool,
+        ) -> PartialVMResult<()>;
+
+        [if is_generic { EXISTS } else { EXISTS_GENERIC }]
+        fn charge_exists(
+            &mut self,
+            is_generic: bool,
+            ty: impl TypeView,
+            exists: bool,
+        ) -> PartialVMResult<()>;
+
+        [if is_generic { MOVE_FROM } else { MOVE_FROM_GENERIC }]
+        fn charge_move_from(
+            &mut self,
+            is_generic: bool,
+            ty: impl TypeView,
+            val: Option<impl ValueView>,
+        ) -> PartialVMResult<()>;
+
+        [if is_generic { MOVE_TO } else { MOVE_TO_GENERIC }]
+        fn charge_move_to(
+            &mut self,
+            is_generic: bool,
+            ty: impl TypeView,
+            val: impl ValueView,
+            is_success: bool,
+        ) -> PartialVMResult<()>;
+
+        [VEC_PACK]
+        fn charge_vec_pack<'a>(
+            &mut self,
+            ty: impl TypeView + 'a,
+            args: impl ExactSizeIterator<Item = impl ValueView> + Clone,
+        ) -> PartialVMResult<()>;
+
+        [VEC_LEN]
+        fn charge_vec_len(&mut self, ty: impl TypeView) -> PartialVMResult<()>;
+
+        [VEC_IMM_BORROW]
+        fn charge_vec_borrow(
+            &mut self,
+            is_mut: bool,
+            ty: impl TypeView,
+            is_success: bool,
+        ) -> PartialVMResult<()>;
+
+        [VEC_PUSH_BACK]
+        fn charge_vec_push_back(
+            &mut self,
+            ty: impl TypeView,
+            val: impl ValueView,
+        ) -> PartialVMResult<()>;
+
+        [VEC_POP_BACK]
+        fn charge_vec_pop_back(
+            &mut self,
+            ty: impl TypeView,
+            val: Option<impl ValueView>,
+        ) -> PartialVMResult<()>;
+
+        [VEC_UNPACK]
+        fn charge_vec_unpack(
+            &mut self,
+            ty: impl TypeView,
+            expect_num_elements: NumArgs,
+            elems: impl ExactSizeIterator<Item = impl ValueView> + Clone,
+        ) -> PartialVMResult<()>;
+
+        [VEC_SWAP]
+        fn charge_vec_swap(&mut self, ty: impl TypeView) -> PartialVMResult<()>;
+    }
+
+    fn balance_internal(&self) -> InternalGas {
+        self.base.balance_internal()
+    }
+
+    fn charge_native_function(
+        &mut self,
+        amount: InternalGas,
+        ret_vals: Option<impl ExactSizeIterator<Item = impl ValueView> + Clone>,
+    ) -> PartialVMResult<()> {
+        let (cost, res) =
+            self.delegate_charge(|base| base.charge_native_function(amount, ret_vals));
+
+        let cur = self.frames.pop().expect("frame must exist");
+        let last = self.frames.last_mut().expect("frame must exist");
+
+        let (module_id, name, ty_args) = match cur.name {
+            FrameName::Function {
+                module_id,
+                name,
+                ty_args,
+            } => (module_id, name, ty_args),
+            FrameName::Script => unreachable!(),
+        };
+
+        last.events.push(ExecutionGasEvent::CallNative {
+            module_id,
+            fn_name: name,
+            ty_args,
+            cost,
+        });
+
+        res
+    }
+
+    fn charge_br_false(&mut self, target_offset: Option<CodeOffset>) -> PartialVMResult<()> {
+        let (cost, res) = self.delegate_charge(|base| base.charge_br_false(target_offset));
+
+        self.record_bytecode(Opcodes::BR_FALSE, cost);
+        if let Some(offset) = target_offset {
+            self.record_offset(offset);
+        }
+
+        res
+    }
+
+    fn charge_br_true(&mut self, target_offset: Option<CodeOffset>) -> PartialVMResult<()> {
+        let (cost, res) = self.delegate_charge(|base| base.charge_br_true(target_offset));
+
+        self.record_bytecode(Opcodes::BR_TRUE, cost);
+        if let Some(offset) = target_offset {
+            self.record_offset(offset);
+        }
+
+        res
+    }
+
+    fn charge_branch(&mut self, target_offset: CodeOffset) -> PartialVMResult<()> {
+        let (cost, res) = self.delegate_charge(|base| base.charge_branch(target_offset));
+
+        self.record_bytecode(Opcodes::BRANCH, cost);
+        self.record_offset(target_offset);
+
+        res
+    }
+
+    fn charge_simple_instr(&mut self, instr: SimpleInstruction) -> PartialVMResult<()> {
+        let (cost, res) = self.delegate_charge(|base| base.charge_simple_instr(instr));
+
+        self.record_bytecode(instr.to_opcode(), cost);
+
+        // TODO: Right now we keep the last frame on the stack even after hitting the ret instruction,
+        //       so that it can be picked up by finishing procedure.
+        //       This is a bit hacky and can lead to weird behaviors if the profiler is used
+        //       over multiple transactions, but again, guarding against that case is a broader
+        //       problem we can deal with in the future.
+        if matches!(instr, SimpleInstruction::Ret) && self.frames.len() > 1 {
+            let cur_frame = self.frames.pop().expect("frame must exist");
+            let last_frame = self.frames.last_mut().expect("frame must exist");
+            last_frame.events.push(ExecutionGasEvent::Call(cur_frame));
+        }
+
+        res
+    }
+
+    fn charge_call(
+        &mut self,
+        module_id: &ModuleId,
+        func_name: &str,
+        args: impl ExactSizeIterator<Item = impl ValueView> + Clone,
+        num_locals: NumArgs,
+    ) -> PartialVMResult<()> {
+        let (cost, res) =
+            self.delegate_charge(|base| base.charge_call(module_id, func_name, args, num_locals));
+
+        self.record_bytecode(Opcodes::CALL, cost);
+        self.frames.push(CallFrame::new_function(
+            module_id.clone(),
+            Identifier::new(func_name).unwrap(),
+            vec![],
+        ));
+
+        res
+    }
+
+    fn charge_call_generic(
+        &mut self,
+        module_id: &ModuleId,
+        func_name: &str,
+        ty_args: impl ExactSizeIterator<Item = impl TypeView> + Clone,
+        args: impl ExactSizeIterator<Item = impl ValueView> + Clone,
+        num_locals: NumArgs,
+    ) -> PartialVMResult<()> {
+        let ty_tags = ty_args
+            .clone()
+            .map(|ty| ty.to_type_tag())
+            .collect::<Vec<_>>();
+
+        let (cost, res) = self.delegate_charge(|base| {
+            base.charge_call_generic(module_id, func_name, ty_args, args, num_locals)
+        });
+
+        self.record_bytecode(Opcodes::CALL_GENERIC, cost);
+        self.frames.push(CallFrame::new_function(
+            module_id.clone(),
+            Identifier::new(func_name).unwrap(),
+            ty_tags,
+        ));
+
+        res
+    }
+
+    fn charge_load_resource(
+        &mut self,
+        addr: AccountAddress,
+        ty: impl TypeView,
+        loaded: Option<(NumBytes, impl ValueView)>,
+    ) -> PartialVMResult<()> {
+        let ty_tag = ty.to_type_tag();
+
+        let (cost, res) = self.delegate_charge(|base| base.charge_load_resource(addr, ty, loaded));
+
+        self.active_event_stream()
+            .push(ExecutionGasEvent::LoadResource {
+                addr,
+                ty: ty_tag,
+                cost,
+            });
+
+        res
+    }
+}
+
+fn write_op_type(op: &WriteOp) -> WriteOpType {
+    use WriteOp as O;
+    use WriteOpType as T;
+
+    match op {
+        O::Creation(..) | O::CreationWithMetadata { .. } => T::Creation,
+        O::Modification(..) | O::ModificationWithMetadata { .. } => T::Modification,
+        O::Deletion | O::DeletionWithMetadata { .. } => T::Deletion,
+    }
+}
+
+impl<G> AptosGasMeter for GasProfiler<G>
+where
+    G: AptosGasMeter,
+{
+    delegate! {
+        fn feature_version(&self) -> u64;
+
+        fn balance(&self) -> Gas;
+
+        fn gas_unit_scaling_factor(&self) -> GasScalingFactor;
+
+        fn io_gas_per_write(&self, key: &StateKey, op: &WriteOp) -> InternalGas;
+
+        fn storage_fee_per_write(&self, key: &StateKey, op: &WriteOp) -> Fee;
+
+        fn storage_fee_per_event(&self, event: &ContractEvent) -> Fee;
+
+        fn storage_discount_for_events(&self, total_cost: Fee) -> Fee;
+
+        fn storage_fee_for_transaction_storage(&self, txn_size: NumBytes) -> Fee;
+    }
+
+    delegate_mut! {
+        fn charge_execution(&mut self, amount: InternalGas) -> PartialVMResult<()>;
+
+        fn charge_io(&mut self, amount: InternalGas) -> PartialVMResult<()>;
+
+        fn charge_storage_fee(
+            &mut self,
+            amount: aptos_gas::Fee,
+            gas_unit_price: aptos_gas::FeePerGasUnit,
+        ) -> PartialVMResult<()>;
+    }
+
+    fn charge_io_gas_for_write_set<'a>(
+        &mut self,
+        ops: impl IntoIterator<Item = (&'a StateKey, &'a WriteOp)>,
+    ) -> VMResult<()> {
+        for (key, op) in ops {
+            let cost = self.io_gas_per_write(key, op);
+            self.write_set_transient.push(WriteTransient {
+                key: key.clone(),
+                cost,
+                op_type: write_op_type(op),
+            });
+            self.charge_io(cost)
+                .map_err(|e| e.finish(Location::Undefined))?;
+        }
+        Ok(())
+    }
+
+    fn charge_storage_fee_for_all<'a>(
+        &mut self,
+        write_ops: impl IntoIterator<Item = (&'a StateKey, &'a WriteOp)>,
+        events: impl IntoIterator<Item = &'a ContractEvent>,
+        txn_size: NumBytes,
+        gas_unit_price: aptos_gas::FeePerGasUnit,
+    ) -> VMResult<()> {
+        // The new storage fee are only active since version 7.
+        if self.feature_version() < 7 {
+            return Ok(());
+        }
+
+        // TODO(Gas): right now, some of our tests use a unit price of 0 and this is a hack
+        // to avoid causing them issues. We should revisit the problem and figure out a
+        // better way to handle this.
+        if gas_unit_price.is_zero() {
+            return Ok(());
+        }
+
+        // Writes
+        let mut write_fee = Fee::new(0);
+        let mut write_set_storage = vec![];
+        for (key, op) in write_ops.into_iter() {
+            let fee = self.storage_fee_per_write(key, op);
+            write_set_storage.push(WriteStorage {
+                key: key.clone(),
+                op_type: write_op_type(op),
+                cost: fee,
+            });
+            write_fee += fee;
+        }
+
+        // Events
+        let mut event_fee = Fee::new(0);
+        let mut event_fees = vec![];
+        for event in events {
+            let fee = self.storage_fee_per_event(event);
+            event_fees.push(EventStorage {
+                ty: event.type_tag().clone(),
+                cost: fee,
+            });
+            event_fee += fee;
+        }
+        let event_discount = self.storage_discount_for_events(event_fee);
+        event_fee = event_fee
+            .checked_sub(event_discount)
+            .expect("discount should always be less than or equal to total amount");
+
+        // Txn
+        let txn_fee = self.storage_fee_for_transaction_storage(txn_size);
+
+        self.storage_fees = Some(StorageFees {
+            write_set_storage,
+            events: event_fees,
+            event_discount,
+            txn_storage: txn_fee,
+        });
+
+        self.charge_storage_fee(write_fee + event_fee + txn_fee, gas_unit_price)
+            .map_err(|err| err.finish(Location::Undefined))?;
+
+        Ok(())
+    }
+
+    fn charge_intrinsic_gas_for_transaction(&mut self, txn_size: NumBytes) -> VMResult<()> {
+        let (cost, res) =
+            self.delegate_charge(|base| base.charge_intrinsic_gas_for_transaction(txn_size));
+
+        self.intrinsic_cost = Some(cost);
+
+        res
+    }
+}
+
+impl<G> GasProfiler<G>
+where
+    G: AptosGasMeter,
+{
+    pub fn finish(mut self) -> TransactionGasLog {
+        while self.frames.len() > 1 {
+            let cur = self.frames.pop().expect("frame must exist");
+            let last = self.frames.last_mut().expect("frame must exist");
+            last.events.push(ExecutionGasEvent::Call(cur));
+        }
+
+        TransactionGasLog {
+            gas_scaling_factor: self.base.gas_unit_scaling_factor(),
+            intrinsic_cost: self.intrinsic_cost.unwrap_or_else(|| 0.into()),
+            call_graph: self.frames.pop().expect("frame must exist"),
+            write_set_transient: self.write_set_transient,
+            storage: self.storage_fees.unwrap_or_else(|| StorageFees {
+                write_set_storage: vec![],
+                events: vec![],
+                event_discount: 0.into(),
+                txn_storage: 0.into(),
+            }),
+        }
+    }
+}
diff --git a/aptos-move/aptos-gas/src/gas_meter.rs b/aptos-move/aptos-gas/src/gas_meter.rs
index dd77cffc46e18..9f1148bc4bd18 100644
--- a/aptos-move/aptos-gas/src/gas_meter.rs
+++ b/aptos-move/aptos-gas/src/gas_meter.rs
@@ -8,8 +8,8 @@ use crate::{
     algebra::{AbstractValueSize, Fee, Gas},
     instr::InstructionGasParameters,
     misc::MiscGasParameters,
-    transaction::{ChangeSetConfigs, TransactionGasParameters},
-    FeePerGasUnit, StorageGasParameters,
+    transaction::TransactionGasParameters,
+    FeePerGasUnit, GasScalingFactor, StorageGasParameters,
 };
 use aptos_logger::error;
 use aptos_types::{
@@ -27,7 +27,7 @@ use move_core_types::{
     vm_status::StatusCode,
 };
 use move_vm_types::{
-    gas::{GasMeter, SimpleInstruction},
+    gas::{GasMeter as MoveGasMeter, SimpleInstruction},
     views::{TypeView, ValueView},
 };
 use std::collections::BTreeMap;
@@ -213,10 +213,127 @@ impl InitialGasSchedule for AptosGasParameters {
     }
 }
 
+/// Trait representing a gas meter used inside the Aptos VM.
+///
+/// It extends Move VM's `GasMeter` trait with a few Aptos-specific callbacks, along with
+/// some primitives that would allow gas meters to be composable.
+pub trait AptosGasMeter: MoveGasMeter {
+    /// Returns the gas feature version.
+    fn feature_version(&self) -> u64;
+
+    /// Returns the remaining balance, measured in (external) gas units.
+    ///
+    /// The number should be rounded down when converting from internal to external gas units.
+    fn balance(&self) -> Gas;
+
+    /// Returns the scaling factor between internal and external gas units.
+    fn gas_unit_scaling_factor(&self) -> GasScalingFactor;
+
+    /// Charges gas for performing operations that fall under the execution category.
+    fn charge_execution(&mut self, amount: InternalGas) -> PartialVMResult<()>;
+
+    /// Charges gas for performing operations that fall under the IO category.
+    fn charge_io(&mut self, amount: InternalGas) -> PartialVMResult<()>;
+
+    /// Charges fee for utlizing short-term or long-term storage.
+    ///
+    /// Since the fee is measured in APT/Octa, it needs to be converted into gas units
+    /// according to the given unit price.
+    fn charge_storage_fee(
+        &mut self,
+        amount: Fee,
+        gas_unit_price: FeePerGasUnit,
+    ) -> PartialVMResult<()>;
+
+    /// Charge an intrinsic cost for executing the transaction.
+    ///
+    /// The cost stays constant for transactions below a certain size, but will grow proportionally
+    /// for bigger ones.
+    fn charge_intrinsic_gas_for_transaction(&mut self, txn_size: NumBytes) -> VMResult<()>;
+
+    /// Calculates the IO gas cost required to perform a write operation.
+    fn io_gas_per_write(&self, key: &StateKey, op: &WriteOp) -> InternalGas;
+
+    /// Charges IO gas for all items in the specified write set.
+    ///
+    /// The contract requires that this function behaviors in a way that is consistent to
+    /// `fn io_gas_per_write`.
+    /// Due to this reason, you should normally not override the default implementation,
+    /// unless you are doing something special, such as injecting additional logging logic.
+    fn charge_io_gas_for_write_set<'a>(
+        &mut self,
+        ops: impl IntoIterator<Item = (&'a StateKey, &'a WriteOp)>,
+    ) -> VMResult<()> {
+        for (key, op) in ops {
+            self.charge_io(self.io_gas_per_write(key, op))
+                .map_err(|e| e.finish(Location::Undefined))?;
+        }
+        Ok(())
+    }
+
+    /// Calculates the storage fee for a write operation.
+    fn storage_fee_per_write(&self, key: &StateKey, op: &WriteOp) -> Fee;
+
+    /// Calculates the storage fee for an event.
+    fn storage_fee_per_event(&self, event: &ContractEvent) -> Fee;
+
+    /// Calculates the discount applied to the event storage fees, based on a free quota.
+    fn storage_discount_for_events(&self, total_cost: Fee) -> Fee;
+
+    /// Calculates the storage fee for the transaction.
+    fn storage_fee_for_transaction_storage(&self, txn_size: NumBytes) -> Fee;
+
+    /// Charges the storage fees for writes, events & txn storage in a lump sum, minimizing the
+    /// loss of precision.
+    ///
+    /// The contract requires that this function behaviors in a way that is consistent to
+    /// the ones defining the costs.
+    /// Due to this reason, you should normally not override the default implementation,
+    /// unless you are doing something special, such as injecting additional logging logic.
+    fn charge_storage_fee_for_all<'a>(
+        &mut self,
+        write_ops: impl IntoIterator<Item = (&'a StateKey, &'a WriteOp)>,
+        events: impl IntoIterator<Item = &'a ContractEvent>,
+        txn_size: NumBytes,
+        gas_unit_price: FeePerGasUnit,
+    ) -> VMResult<()> {
+        // The new storage fee are only active since version 7.
+        if self.feature_version() < 7 {
+            return Ok(());
+        }
+
+        // TODO(Gas): right now, some of our tests use a unit price of 0 and this is a hack
+        // to avoid causing them issues. We should revisit the problem and figure out a
+        // better way to handle this.
+        if gas_unit_price.is_zero() {
+            return Ok(());
+        }
+
+        // Calculate the storage fees.
+        let write_fee = write_ops.into_iter().fold(Fee::new(0), |acc, (key, op)| {
+            acc + self.storage_fee_per_write(key, op)
+        });
+        let event_fee = events.into_iter().fold(Fee::new(0), |acc, event| {
+            acc + self.storage_fee_per_event(event)
+        });
+        let event_discount = self.storage_discount_for_events(event_fee);
+        let event_net_fee = event_fee
+            .checked_sub(event_discount)
+            .expect("discount should always be less than or equal to total amount");
+        let txn_fee = self.storage_fee_for_transaction_storage(txn_size);
+        let fee = write_fee + event_net_fee + txn_fee;
+
+        self.charge_storage_fee(fee, gas_unit_price)
+            .map_err(|err| err.finish(Location::Undefined))?;
+
+        Ok(())
+    }
+}
+
 /// The official gas meter used inside the Aptos VM.
 /// It maintains an internal gas counter, measured in internal gas units, and carries an environment
 /// consisting all the gas parameters, which it can lookup when performing gas calculations.
-pub struct AptosGasMeter {
+pub struct StandardGasMeter {
     feature_version: u64,
     gas_params: AptosGasParameters,
     storage_gas_params: StorageGasParameters,
@@ -230,7 +347,7 @@ pub struct AptosGasMeter {
     should_leak_memory_for_native: bool,
 }
 
-impl AptosGasMeter {
+impl StandardGasMeter {
     pub fn new(
         gas_feature_version: u64,
         gas_params: AptosGasParameters,
@@ -253,11 +370,6 @@ impl AptosGasMeter {
         }
     }
 
-    pub fn balance(&self) -> Gas {
-        self.balance
-            .to_unit_round_down_with_params(&self.gas_params.txn)
-    }
-
     #[inline]
     fn charge(&mut self, amount: InternalGas) -> PartialVMResult<()> {
         match self.balance.checked_sub(amount) {
@@ -272,32 +384,6 @@ impl AptosGasMeter {
         }
     }
 
-    #[inline]
-    fn charge_execution(&mut self, amount: InternalGas) -> PartialVMResult<()> {
-        self.charge(amount)?;
-
-        self.execution_gas_used += amount;
-        if self.feature_version >= 7
-            && self.execution_gas_used > self.gas_params.txn.max_execution_gas
-        {
-            Err(PartialVMError::new(StatusCode::EXECUTION_LIMIT_REACHED))
-        } else {
-            Ok(())
-        }
-    }
-
-    #[inline]
-    fn charge_io(&mut self, amount: InternalGas) -> PartialVMResult<()> {
-        self.charge(amount)?;
-
-        self.io_gas_used += amount;
-        if self.feature_version >= 7 && self.io_gas_used > self.gas_params.txn.max_io_gas {
-            Err(PartialVMError::new(StatusCode::IO_LIMIT_REACHED))
-        } else {
-            Ok(())
-        }
-    }
-
     #[inline]
     fn use_heap_memory(&mut self, amount: AbstractValueSize) -> PartialVMResult<()> {
         if self.feature_version >= 3 {
@@ -326,25 +412,25 @@ impl AptosGasMeter {
     pub fn feature_version(&self) -> u64 {
         self.feature_version
     }
-
-    pub fn change_set_configs(&self) -> &ChangeSetConfigs {
-        &self.storage_gas_params.change_set_configs
-    }
 }
 
-impl GasMeter for AptosGasMeter {
+impl MoveGasMeter for StandardGasMeter {
+    #[inline]
     fn balance_internal(&self) -> InternalGas {
         self.balance
     }
 
+    #[inline]
     fn charge_br_false(&mut self, _target_offset: Option<CodeOffset>) -> PartialVMResult<()> {
         self.charge_execution(self.gas_params.instr.br_false)
     }
 
+    #[inline]
     fn charge_br_true(&mut self, _target_offset: Option<CodeOffset>) -> PartialVMResult<()> {
         self.charge_execution(self.gas_params.instr.br_true)
     }
 
+    #[inline]
     fn charge_branch(&mut self, _target_offset: CodeOffset) -> PartialVMResult<()> {
         self.charge_execution(self.gas_params.instr.branch)
     }
@@ -826,47 +912,53 @@ impl GasMeter for AptosGasMeter {
     }
 }
 
-impl AptosGasMeter {
-    pub fn charge_intrinsic_gas_for_transaction(&mut self, txn_size: NumBytes) -> VMResult<()> {
-        let cost = self.gas_params.txn.calculate_intrinsic_gas(txn_size);
-        self.charge_execution(cost)
-            .map_err(|e| e.finish(Location::Undefined))
+impl AptosGasMeter for StandardGasMeter {
+    fn feature_version(&self) -> u64 {
+        self.feature_version
     }
 
-    pub fn charge_write_set_gas_for_io<'a>(
-        &mut self,
-        ops: impl IntoIterator<Item = (&'a StateKey, &'a WriteOp)>,
-    ) -> VMResult<()> {
-        let cost = self.storage_gas_params.pricing.calculate_write_set_gas(ops);
-        self.charge_io(cost)
-            .map_err(|e| e.finish(Location::Undefined))
+    fn balance(&self) -> Gas {
+        self.balance
+            .to_unit_round_down_with_params(&self.gas_params.txn)
     }
 
-    pub fn charge_storage_fee<'a>(
-        &mut self,
-        write_ops: impl IntoIterator<Item = (&'a StateKey, &'a WriteOp)>,
-        events: impl IntoIterator<Item = &'a ContractEvent>,
-        txn_size: NumBytes,
-        gas_unit_price: FeePerGasUnit,
-    ) -> VMResult<()> {
-        // The new storage fee are only active since version 7.
-        if self.feature_version < 7 {
-            return Ok(());
+    fn gas_unit_scaling_factor(&self) -> GasScalingFactor {
+        self.gas_params.txn.gas_unit_scaling_factor
+    }
+
+    #[inline]
+    fn charge_execution(&mut self, amount: InternalGas) -> PartialVMResult<()> {
+        self.charge(amount)?;
+
+        self.execution_gas_used += amount;
+        if self.feature_version >= 7
+            && self.execution_gas_used > self.gas_params.txn.max_execution_gas
+        {
+            Err(PartialVMError::new(StatusCode::EXECUTION_LIMIT_REACHED))
+        } else {
+            Ok(())
         }
+    }
 
-        // TODO(Gas): right now, some of our tests use a unit price of 0 and this is a hack
-        // to avoid causing them issues. We should revisit the problem and figure out a
-        // better way to handle this.
-        if gas_unit_price.is_zero() {
-            return Ok(());
+    #[inline]
+    fn charge_io(&mut self, amount: InternalGas) -> PartialVMResult<()> {
+        self.charge(amount)?;
+
+        self.io_gas_used += amount;
+        if self.feature_version >= 7 && self.io_gas_used > self.gas_params.txn.max_io_gas {
+            Err(PartialVMError::new(StatusCode::IO_LIMIT_REACHED))
+        } else {
+            Ok(())
         }
+    }
 
-        // Calculate the storage fees.
+    #[inline]
+    fn charge_storage_fee(
+        &mut self,
+        amount: Fee,
+        gas_unit_price: FeePerGasUnit,
+    ) -> PartialVMResult<()> {
         let txn_params = &self.gas_params.txn;
-        let write_fee = txn_params.calculate_write_set_storage_fee(write_ops);
-        let event_fee = txn_params.calculate_event_storage_fee(events);
-        let txn_fee = txn_params.calculate_transaction_storage_fee(txn_size);
-        let fee = write_fee + event_fee + txn_fee;
 
         // Because the storage fees are defined in terms of fixed APT costs, we need
         // to convert them into gas units.
@@ -881,7 +973,7 @@ impl AptosGasMeter {
             }
         }
         let gas_consumed_internal = div_ceil(
-            (u64::from(fee) as u128) * (u64::from(txn_params.gas_unit_scaling_factor) as u128),
+            (u64::from(amount) as u128) * (u64::from(txn_params.gas_unit_scaling_factor) as u128),
             u64::from(gas_unit_price) as u128,
         );
         let gas_consumed_internal = InternalGas::new(
@@ -896,17 +988,42 @@ impl AptosGasMeter {
             },
         );
 
-        self.charge(gas_consumed_internal)
-            .map_err(|e| e.finish(Location::Undefined))?;
+        self.charge(gas_consumed_internal)?;
 
-        self.storage_fee_used += fee;
+        self.storage_fee_used += amount;
         if self.feature_version >= 7 && self.storage_fee_used > self.gas_params.txn.max_storage_fee
         {
-            return Err(
-                PartialVMError::new(StatusCode::STORAGE_LIMIT_REACHED).finish(Location::Undefined)
-            );
+            return Err(PartialVMError::new(StatusCode::STORAGE_LIMIT_REACHED));
         }
 
         Ok(())
     }
+
+    fn io_gas_per_write(&self, key: &StateKey, op: &WriteOp) -> InternalGas {
+        self.storage_gas_params.pricing.io_gas_per_write(key, op)
+    }
+
+    fn storage_fee_per_write(&self, key: &StateKey, op: &WriteOp) -> Fee {
+        self.gas_params.txn.storage_fee_per_write(key, op)
+    }
+
+    fn storage_fee_per_event(&self, event: &ContractEvent) -> Fee {
+        self.gas_params.txn.storage_fee_per_event(event)
+    }
+
+    fn storage_discount_for_events(&self, total_cost: Fee) -> Fee {
+        self.gas_params.txn.storage_discount_for_events(total_cost)
+    }
+
+    fn storage_fee_for_transaction_storage(&self, txn_size: NumBytes) -> Fee {
+        self.gas_params
+            .txn
+            .storage_fee_for_transaction_storage(txn_size)
+    }
+
+    fn charge_intrinsic_gas_for_transaction(&mut self, txn_size: NumBytes) -> VMResult<()> {
+        let cost = self.gas_params.txn.calculate_intrinsic_gas(txn_size);
+        self.charge_execution(cost)
+            .map_err(|e| e.finish(Location::Undefined))
+    }
 }
diff --git a/aptos-move/aptos-gas/src/instr.rs b/aptos-move/aptos-gas/src/instr.rs
index 070450d75da37..aeea7f2ab09b7 100644
--- a/aptos-move/aptos-gas/src/instr.rs
+++ b/aptos-move/aptos-gas/src/instr.rs
@@ -218,11 +218,6 @@ impl InstructionGasParameters {
             Abort => self.abort,
             Ret => self.ret,
 
-            /* TODO: fixme
-            BrTrue => self.br_true,
-            BrFalse => self.br_false,
-            Branch => self.branch,
-            */
             LdU8 => self.ld_u8,
             LdU16 => self.ld_u16,
             LdU32 => self.ld_u32,
diff --git a/aptos-move/aptos-gas/src/lib.rs b/aptos-move/aptos-gas/src/lib.rs
index aea85115d5b0f..5961345c1c867 100644
--- a/aptos-move/aptos-gas/src/lib.rs
+++ b/aptos-move/aptos-gas/src/lib.rs
@@ -37,7 +37,7 @@ mod transaction;
 pub use algebra::*;
 pub use gas_meter::{
     AptosGasMeter, AptosGasParameters, FromOnChainGasSchedule, InitialGasSchedule,
-    NativeGasParameters, ToOnChainGasSchedule, LATEST_GAS_FEATURE_VERSION,
+    NativeGasParameters, StandardGasMeter, ToOnChainGasSchedule, LATEST_GAS_FEATURE_VERSION,
 };
 pub use instr::InstructionGasParameters;
 pub use misc::{AbstractValueSizeGasParameters, MiscGasParameters};
diff --git a/aptos-move/aptos-gas/src/transaction/mod.rs b/aptos-move/aptos-gas/src/transaction/mod.rs
index f411b1ea29ce8..99db9c8d8944e 100644
--- a/aptos-move/aptos-gas/src/transaction/mod.rs
+++ b/aptos-move/aptos-gas/src/transaction/mod.rs
@@ -186,60 +186,41 @@ impl TransactionGasParameters {
         }
     }
 
-    /// New formula to charge storage fee for write set items based on fixed APT costs.
-    pub fn calculate_write_set_storage_fee<'a>(
-        &self,
-        ops: impl IntoIterator<Item = (&'a StateKey, &'a WriteOp)>,
-    ) -> Fee {
+    /// New formula to charge storage fee for a write, measured in APT.
+    pub fn storage_fee_per_write(&self, key: &StateKey, op: &WriteOp) -> Fee {
         use WriteOp::*;
 
-        let excess_fee = |size: NumBytes| -> Fee {
+        let excess_fee = |key: &StateKey, data: &[u8]| -> Fee {
+            let size = NumBytes::new(key.size() as u64) + NumBytes::new(data.len() as u64);
             match size.checked_sub(self.free_write_bytes_quota) {
                 Some(excess) => excess * self.storage_fee_per_excess_state_byte,
                 None => 0.into(),
             }
         };
 
-        let mut fee = Fee::zero();
-        let mut new_slots = NumSlots::zero();
-
-        for (key, op) in ops {
-            let key_size = NumBytes::new(key.size() as u64);
-            match op {
-                Creation(data) | CreationWithMetadata { data, .. } => {
-                    new_slots += 1.into();
-                    let val_size = NumBytes::new(data.len() as u64);
-                    fee += excess_fee(key_size + val_size);
-                },
-                Modification(data) | ModificationWithMetadata { data, .. } => {
-                    let val_size = NumBytes::new(data.len() as u64);
-                    fee += excess_fee(key_size + val_size);
-                },
-                Deletion | DeletionWithMetadata { .. } => (),
-            }
+        match op {
+            Creation(data) | CreationWithMetadata { data, .. } => {
+                self.storage_fee_per_state_slot_create * NumSlots::new(1) + excess_fee(key, data)
+            },
+            Modification(data) | ModificationWithMetadata { data, .. } => excess_fee(key, data),
+            Deletion | DeletionWithMetadata { .. } => 0.into(),
         }
+    }
 
-        fee += new_slots * self.storage_fee_per_state_slot_create;
-
-        fee
+    /// New formula to charge storage fee for an event, measured in APT.
+    pub fn storage_fee_per_event(&self, event: &ContractEvent) -> Fee {
+        NumBytes::new(event.size() as u64) * self.storage_fee_per_event_byte
     }
 
-    /// New formula to charge storage fee for events based on fixed APT costs.
-    pub fn calculate_event_storage_fee<'a>(
-        &self,
-        events: impl IntoIterator<Item = &'a ContractEvent>,
-    ) -> Fee {
-        let total_bytes = events.into_iter().fold(NumBytes::zero(), |total, event| {
-            total + NumBytes::new(event.size() as u64)
-        });
-        total_bytes
-            .checked_sub(self.free_event_bytes_quota)
-            .unwrap_or(NumBytes::zero())
-            * self.storage_fee_per_event_byte
+    pub fn storage_discount_for_events(&self, total_cost: Fee) -> Fee {
+        std::cmp::min(
+            total_cost,
+            self.free_event_bytes_quota * self.storage_fee_per_event_byte,
+        )
     }
 
-    /// New formula to charge storage fee for transaction based on fixed APT costs.
-    pub fn calculate_transaction_storage_fee(&self, txn_size: NumBytes) -> Fee {
+    /// New formula to charge storage fee for transaction, measured in APT.
+    pub fn storage_fee_for_transaction_storage(&self, txn_size: NumBytes) -> Fee {
         txn_size
             .checked_sub(self.large_transaction_cutoff)
             .unwrap_or(NumBytes::zero())
diff --git a/aptos-move/aptos-gas/src/transaction/storage.rs b/aptos-move/aptos-gas/src/transaction/storage.rs
index 7582c5e2034cf..e056a45efc7f5 100644
--- a/aptos-move/aptos-gas/src/transaction/storage.rs
+++ b/aptos-move/aptos-gas/src/transaction/storage.rs
@@ -48,49 +48,32 @@ impl StoragePricingV1 {
             }
     }
 
-    fn calculate_write_set_gas<'a>(
-        &self,
-        ops: impl IntoIterator<Item = (&'a StateKey, &'a WriteOp)>,
-    ) -> InternalGas {
-        use WriteOp::*;
-
-        // Counting
-        let mut num_ops = NumArgs::zero();
-        let mut num_new_items = NumArgs::zero();
-        let mut num_bytes_key = NumBytes::zero();
-        let mut num_bytes_val = NumBytes::zero();
-
-        for (key, op) in ops {
-            num_ops += 1.into();
-
-            if self.write_data_per_byte_in_key > 0.into() {
-                // TODO(Gas): Are we supposed to panic here?
-                num_bytes_key += NumBytes::new(
+    fn io_gas_per_write(&self, key: &StateKey, op: &WriteOp) -> InternalGas {
+        use aptos_types::write_set::WriteOp::*;
+
+        let mut cost = self.write_data_per_op * NumArgs::new(1);
+
+        if self.write_data_per_byte_in_key > 0.into() {
+            cost += self.write_data_per_byte_in_key
+                * NumBytes::new(
                     key.encode()
                         .expect("Should be able to serialize state key")
                         .len() as u64,
                 );
-            }
-
-            match op {
-                Creation(data) | CreationWithMetadata { data, .. } => {
-                    num_new_items += 1.into();
-                    num_bytes_val += NumBytes::new(data.len() as u64);
-                },
-                Modification(data) | ModificationWithMetadata { data, .. } => {
-                    num_bytes_val += NumBytes::new(data.len() as u64);
-                },
-                Deletion | DeletionWithMetadata { .. } => (),
-            }
         }
 
-        // Calculate the costs
-        let cost_ops = self.write_data_per_op * num_ops;
-        let cost_new_items = self.write_data_per_new_item * num_new_items;
-        let cost_bytes = self.write_data_per_byte_in_key * num_bytes_key
-            + self.write_data_per_byte_in_val * num_bytes_val;
+        match op {
+            Creation(data) | CreationWithMetadata { data, .. } => {
+                cost += self.write_data_per_new_item * NumArgs::new(1)
+                    + self.write_data_per_byte_in_val * NumBytes::new(data.len() as u64);
+            },
+            Modification(data) | ModificationWithMetadata { data, .. } => {
+                cost += self.write_data_per_byte_in_val * NumBytes::new(data.len() as u64);
+            },
+            Deletion | DeletionWithMetadata { .. } => (),
+        }
 
-        cost_ops + cost_new_items + cost_bytes
+        cost
     }
 }
 
@@ -170,35 +153,20 @@ impl StoragePricingV2 {
             }
     }
 
-    fn calculate_write_set_gas<'a>(
-        &self,
-        ops: impl IntoIterator<Item = (&'a StateKey, &'a WriteOp)>,
-    ) -> InternalGas {
+    fn io_gas_per_write(&self, key: &StateKey, op: &WriteOp) -> InternalGas {
         use aptos_types::write_set::WriteOp::*;
 
-        let mut num_items_create = NumArgs::zero();
-        let mut num_items_write = NumArgs::zero();
-        let mut num_bytes_create = NumBytes::zero();
-        let mut num_bytes_write = NumBytes::zero();
-
-        for (key, op) in ops {
-            match &op {
-                Creation(data) | CreationWithMetadata { data, .. } => {
-                    num_items_create += 1.into();
-                    num_bytes_create += self.write_op_size(key, data);
-                },
-                Modification(data) | ModificationWithMetadata { data, .. } => {
-                    num_items_write += 1.into();
-                    num_bytes_write += self.write_op_size(key, data);
-                },
-                Deletion | DeletionWithMetadata { .. } => (),
-            }
+        match &op {
+            Creation(data) | CreationWithMetadata { data, .. } => {
+                self.per_item_create * NumArgs::new(1)
+                    + self.write_op_size(key, data) * self.per_byte_create
+            },
+            Modification(data) | ModificationWithMetadata { data, .. } => {
+                self.per_item_write * NumArgs::new(1)
+                    + self.write_op_size(key, data) * self.per_byte_write
+            },
+            Deletion | DeletionWithMetadata { .. } => 0.into(),
         }
-
-        num_items_create * self.per_item_create
-            + num_items_write * self.per_item_write
-            + num_bytes_create * self.per_byte_create
-            + num_bytes_write * self.per_byte_write
     }
 }
 
@@ -218,15 +186,12 @@ impl StoragePricing {
         }
     }
 
-    pub fn calculate_write_set_gas<'a>(
-        &self,
-        ops: impl IntoIterator<Item = (&'a StateKey, &'a WriteOp)>,
-    ) -> InternalGas {
+    pub fn io_gas_per_write(&self, key: &StateKey, op: &WriteOp) -> InternalGas {
         use StoragePricing::*;
 
         match self {
-            V1(v1) => v1.calculate_write_set_gas(&mut ops.into_iter()),
-            V2(v2) => v2.calculate_write_set_gas(&mut ops.into_iter()),
+            V1(v1) => v1.io_gas_per_write(key, op),
+            V2(v2) => v2.io_gas_per_write(key, op),
         }
     }
 }
diff --git a/aptos-move/aptos-vm/Cargo.toml b/aptos-move/aptos-vm/Cargo.toml
index 64f1d27ce079f..b415e8df8f859 100644
--- a/aptos-move/aptos-vm/Cargo.toml
+++ b/aptos-move/aptos-vm/Cargo.toml
@@ -58,3 +58,4 @@ mirai-contracts = []
 fuzzing = ["move-core-types/fuzzing", "move-binary-format/fuzzing", "move-vm-types/fuzzing", "aptos-framework/fuzzing"]
 failpoints = ["fail/failpoints", "move-vm-runtime/failpoints"]
 testing = ["move-unit-test", "aptos-framework/testing"]
+ 
diff --git a/aptos-move/aptos-vm/src/aptos_vm.rs b/aptos-move/aptos-vm/src/aptos_vm.rs
index 63ac575edbdc4..a34627d54ebd5 100644
--- a/aptos-move/aptos-vm/src/aptos_vm.rs
+++ b/aptos-move/aptos-vm/src/aptos_vm.rs
@@ -9,7 +9,7 @@ use crate::{
     aptos_vm_impl::{get_transaction_output, AptosVMImpl, AptosVMInternals},
     block_executor::BlockAptosVM,
     counters::*,
-    data_cache::{AsMoveResolver, IntoMoveResolver},
+    data_cache::{AsMoveResolver, IntoMoveResolver, StorageAdapter},
     delta_state_view::DeltaStateView,
     errors::expect_only_successful_execution,
     move_vm_ext::{MoveResolverExt, SessionExt, SessionId},
@@ -24,7 +24,10 @@ use aptos_aggregator::{
 };
 use aptos_crypto::HashValue;
 use aptos_framework::natives::code::PublishRequest;
-use aptos_gas::{AptosGasMeter, ChangeSetConfigs};
+use aptos_gas::{
+    AptosGasMeter, AptosGasParameters, ChangeSetConfigs, Gas, StandardGasMeter,
+    StorageGasParameters,
+};
 use aptos_logger::prelude::*;
 use aptos_state_view::StateView;
 use aptos_types::{
@@ -41,7 +44,7 @@ use aptos_types::{
     vm_status::{AbortLocation, DiscardedVMStatus, StatusCode, VMStatus},
     write_set::WriteSet,
 };
-use aptos_vm_logging::log_schema::AdapterLogSchema;
+use aptos_vm_logging::{init_speculative_logs, log_schema::AdapterLogSchema};
 use fail::fail_point;
 use move_binary_format::{
     access::ModuleAccess,
@@ -89,6 +92,15 @@ pub struct AptosVM(pub(crate) AptosVMImpl);
 
 struct AptosSimulationVM(AptosVM);
 
+macro_rules! unwrap_or_discard {
+    ($res:expr) => {
+        match $res {
+            Ok(s) => s,
+            Err(e) => return discard_error_vm_status(e),
+        }
+    };
+}
+
 impl AptosVM {
     pub fn new<S: StateView>(state: &S) -> Self {
         Self(AptosVMImpl::new(state))
@@ -111,6 +123,8 @@ impl AptosVM {
 
     /// Get the concurrency level if already set, otherwise return default 1
     /// (sequential execution).
+    ///
+    /// The concurrency level is fixed to 1 if gas profiling is enabled.
     pub fn get_concurrency_level() -> usize {
         match EXECUTION_CONCURRENCY_LEVEL.get() {
             Some(concurrency_level) => *concurrency_level,
@@ -190,10 +204,11 @@ impl AptosVM {
     pub fn failed_transaction_cleanup<S: MoveResolverExt>(
         &self,
         error_code: VMStatus,
-        gas_meter: &mut AptosGasMeter,
+        gas_meter: &mut impl AptosGasMeter,
         txn_data: &TransactionMetadata,
         storage: &S,
         log_context: &AdapterLogSchema,
+        change_set_configs: &ChangeSetConfigs,
     ) -> TransactionOutputExt {
         self.failed_transaction_cleanup_and_keep_vm_status(
             error_code,
@@ -201,6 +216,7 @@ impl AptosVM {
             txn_data,
             storage,
             log_context,
+            change_set_configs,
         )
         .1
     }
@@ -208,10 +224,11 @@ impl AptosVM {
     fn failed_transaction_cleanup_and_keep_vm_status<S: MoveResolverExt>(
         &self,
         error_code: VMStatus,
-        gas_meter: &mut AptosGasMeter,
+        gas_meter: &mut impl AptosGasMeter,
         txn_data: &TransactionMetadata,
         storage: &S,
         log_context: &AdapterLogSchema,
+        change_set_configs: &ChangeSetConfigs,
     ) -> (VMStatus, TransactionOutputExt) {
         let resolver = self.0.new_move_resolver(storage);
         let mut session = self.0.new_session(&resolver, SessionId::txn_meta(txn_data));
@@ -254,7 +271,7 @@ impl AptosVM {
                     gas_meter.balance(),
                     txn_data,
                     status,
-                    gas_meter.change_set_configs(),
+                    change_set_configs,
                 )
                 .unwrap_or_else(|e| discard_error_vm_status(e).1);
                 (error_code, txn_output)
@@ -270,9 +287,10 @@ impl AptosVM {
         &self,
         storage: &S,
         user_txn_change_set_ext: ChangeSetExt,
-        gas_meter: &mut AptosGasMeter,
+        gas_meter: &mut impl AptosGasMeter,
         txn_data: &TransactionMetadata,
         log_context: &AdapterLogSchema,
+        change_set_configs: &ChangeSetConfigs,
     ) -> Result<(VMStatus, TransactionOutputExt), VMStatus> {
         let storage_with_changes =
             DeltaStateView::new(storage, user_txn_change_set_ext.write_set());
@@ -302,7 +320,7 @@ impl AptosVM {
             .run_success_epilogue(&mut session, gas_meter.balance(), txn_data, log_context)?;
 
         let epilogue_change_set_ext = session
-            .finish(&mut (), gas_meter.change_set_configs())
+            .finish(&mut (), change_set_configs)
             .map_err(|e| e.into_vm_status())?;
         let change_set_ext = user_txn_change_set_ext
             .squash(epilogue_change_set_ext)
@@ -332,7 +350,7 @@ impl AptosVM {
     fn validate_and_execute_entry_function<SS: MoveResolverExt>(
         &self,
         session: &mut SessionExt<SS>,
-        gas_meter: &mut AptosGasMeter,
+        gas_meter: &mut impl AptosGasMeter,
         senders: Vec<AccountAddress>,
         script_fn: &EntryFunction,
     ) -> Result<SerializedReturnValues, VMStatus> {
@@ -367,11 +385,12 @@ impl AptosVM {
         &self,
         storage: &S,
         mut session: SessionExt<SS>,
-        gas_meter: &mut AptosGasMeter,
+        gas_meter: &mut impl AptosGasMeter,
         txn_data: &TransactionMetadata,
         payload: &TransactionPayload,
         log_context: &AdapterLogSchema,
         new_published_modules_loaded: &mut bool,
+        change_set_configs: &ChangeSetConfigs,
     ) -> Result<(VMStatus, TransactionOutputExt), VMStatus> {
         fail_point!("move_adapter::execute_script_or_entry_function", |_| {
             Err(VMStatus::Error(
@@ -432,10 +451,10 @@ impl AptosVM {
             )?;
 
             let change_set_ext = session
-                .finish(&mut (), gas_meter.change_set_configs())
+                .finish(&mut (), change_set_configs)
                 .map_err(|e| e.into_vm_status())?;
-            gas_meter.charge_write_set_gas_for_io(change_set_ext.write_set().iter())?;
-            gas_meter.charge_storage_fee(
+            gas_meter.charge_io_gas_for_write_set(change_set_ext.write_set().iter())?;
+            gas_meter.charge_storage_fee_for_all(
                 change_set_ext.write_set().iter(),
                 change_set_ext.change_set().events(),
                 txn_data.transaction_size,
@@ -449,6 +468,7 @@ impl AptosVM {
                 gas_meter,
                 txn_data,
                 log_context,
+                change_set_configs,
             )
         }
     }
@@ -464,11 +484,12 @@ impl AptosVM {
         &self,
         storage: &S,
         mut session: SessionExt<SS>,
-        gas_meter: &mut AptosGasMeter,
+        gas_meter: &mut impl AptosGasMeter,
         txn_data: &TransactionMetadata,
         txn_payload: &Multisig,
         log_context: &AdapterLogSchema,
         new_published_modules_loaded: &mut bool,
+        change_set_configs: &ChangeSetConfigs,
     ) -> Result<(VMStatus, TransactionOutputExt), VMStatus> {
         fail_point!("move_adapter::execute_multisig_transaction", |_| {
             Err(VMStatus::Error(
@@ -562,10 +583,10 @@ impl AptosVM {
             };
             self.failure_multisig_payload_cleanup(
                 storage,
-                gas_meter,
                 execution_error,
                 txn_data,
                 cleanup_args,
+                change_set_configs,
             )?
         } else {
             self.success_multisig_payload_cleanup(
@@ -574,6 +595,7 @@ impl AptosVM {
                 gas_meter,
                 txn_data,
                 cleanup_args,
+                change_set_configs,
             )?
         };
 
@@ -584,13 +606,14 @@ impl AptosVM {
             gas_meter,
             txn_data,
             log_context,
+            change_set_configs,
         )
     }
 
     fn execute_multisig_entry_function<SS: MoveResolverExt>(
         &self,
         session: &mut SessionExt<SS>,
-        gas_meter: &mut AptosGasMeter,
+        gas_meter: &mut impl AptosGasMeter,
         multisig_address: AccountAddress,
         payload: &EntryFunction,
         new_published_modules_loaded: &mut bool,
@@ -614,18 +637,19 @@ impl AptosVM {
         &self,
         storage: &S,
         session: SessionExt<SS>,
-        gas_meter: &mut AptosGasMeter,
+        gas_meter: &mut impl AptosGasMeter,
         txn_data: &TransactionMetadata,
         cleanup_args: Vec<Vec<u8>>,
+        change_set_configs: &ChangeSetConfigs,
     ) -> Result<ChangeSetExt, VMStatus> {
         // Charge gas for writeset before we do cleanup. This ensures we don't charge gas for
         // cleanup writeset changes, which is consistent with outer-level success cleanup
         // flow. We also wouldn't need to worry that we run out of gas when doing cleanup.
         let inner_function_change_set_ext = session
-            .finish(&mut (), gas_meter.change_set_configs())
+            .finish(&mut (), change_set_configs)
             .map_err(|e| e.into_vm_status())?;
-        gas_meter.charge_write_set_gas_for_io(inner_function_change_set_ext.write_set().iter())?;
-        gas_meter.charge_storage_fee(
+        gas_meter.charge_io_gas_for_write_set(inner_function_change_set_ext.write_set().iter())?;
+        gas_meter.charge_storage_fee_for_all(
             inner_function_change_set_ext.write_set().iter(),
             inner_function_change_set_ext.change_set().events(),
             txn_data.transaction_size,
@@ -654,7 +678,7 @@ impl AptosVM {
             &mut UnmeteredGasMeter,
         )?;
         let cleanup_change_set_ext = cleanup_session
-            .finish(&mut (), gas_meter.change_set_configs())
+            .finish(&mut (), change_set_configs)
             .map_err(|e| e.into_vm_status())?;
         // Merge the inner function writeset with cleanup writeset.
         inner_function_change_set_ext
@@ -665,10 +689,10 @@ impl AptosVM {
     fn failure_multisig_payload_cleanup<S: MoveResolverExt + StateView>(
         &self,
         storage: &S,
-        gas_meter: &mut AptosGasMeter,
         execution_error: VMStatus,
         txn_data: &TransactionMetadata,
         mut cleanup_args: Vec<Vec<u8>>,
+        change_set_configs: &ChangeSetConfigs,
     ) -> Result<ChangeSetExt, VMStatus> {
         // Start a fresh session for running cleanup that does not contain any changes from
         // the inner function call earlier (since it failed).
@@ -688,7 +712,7 @@ impl AptosVM {
             &mut UnmeteredGasMeter,
         )?;
         cleanup_session
-            .finish(&mut (), gas_meter.change_set_configs())
+            .finish(&mut (), change_set_configs)
             .map_err(|e| e.into_vm_status())
     }
 
@@ -723,7 +747,7 @@ impl AptosVM {
     fn execute_module_initialization<S: MoveResolverExt>(
         &self,
         session: &mut SessionExt<S>,
-        gas_meter: &mut AptosGasMeter,
+        gas_meter: &mut impl AptosGasMeter,
         modules: &[CompiledModule],
         exists: BTreeSet<ModuleId>,
         senders: &[AccountAddress],
@@ -797,11 +821,12 @@ impl AptosVM {
         &self,
         storage: &S,
         mut session: SessionExt<SS>,
-        gas_meter: &mut AptosGasMeter,
+        gas_meter: &mut impl AptosGasMeter,
         txn_data: &TransactionMetadata,
         modules: &ModuleBundle,
         log_context: &AdapterLogSchema,
         new_published_modules_loaded: &mut bool,
+        change_set_configs: &ChangeSetConfigs,
     ) -> Result<(VMStatus, TransactionOutputExt), VMStatus> {
         if MODULE_BUNDLE_DISALLOWED.load(Ordering::Relaxed) {
             return Err(VMStatus::Error(StatusCode::FEATURE_UNDER_GATING, None));
@@ -845,10 +870,10 @@ impl AptosVM {
         )?;
 
         let change_set_ext = session
-            .finish(&mut (), gas_meter.change_set_configs())
+            .finish(&mut (), change_set_configs)
             .map_err(|e| e.into_vm_status())?;
-        gas_meter.charge_write_set_gas_for_io(change_set_ext.write_set().iter())?;
-        gas_meter.charge_storage_fee(
+        gas_meter.charge_io_gas_for_write_set(change_set_ext.write_set().iter())?;
+        gas_meter.charge_storage_fee_for_all(
             change_set_ext.write_set().iter(),
             change_set_ext.change_set().events(),
             txn_data.transaction_size,
@@ -856,14 +881,21 @@ impl AptosVM {
         )?;
         // TODO(Gas): Charge for aggregator writes
 
-        self.success_transaction_cleanup(storage, change_set_ext, gas_meter, txn_data, log_context)
+        self.success_transaction_cleanup(
+            storage,
+            change_set_ext,
+            gas_meter,
+            txn_data,
+            log_context,
+            change_set_configs,
+        )
     }
 
     /// Resolve a pending code publish request registered via the NativeCodeContext.
     fn resolve_pending_code_publish<S: MoveResolverExt>(
         &self,
         session: &mut SessionExt<S>,
-        gas_meter: &mut AptosGasMeter,
+        gas_meter: &mut impl AptosGasMeter,
         new_published_modules_loaded: &mut bool,
     ) -> VMResult<()> {
         if let Some(PublishRequest {
@@ -974,21 +1006,30 @@ impl AptosVM {
             .finish(Location::Undefined)
     }
 
-    pub(crate) fn execute_user_transaction<S: MoveResolverExt>(
+    fn make_standard_gas_meter(
+        &self,
+        balance: Gas,
+        log_context: &AdapterLogSchema,
+    ) -> Result<StandardGasMeter, VMStatus> {
+        Ok(StandardGasMeter::new(
+            self.0.get_gas_feature_version(),
+            self.0.get_gas_parameters(log_context)?.clone(),
+            self.0.get_storage_gas_parameters(log_context)?.clone(),
+            balance,
+        ))
+    }
+
+    fn execute_user_transaction_impl<S, G>(
         &self,
         storage: &S,
         txn: &SignatureCheckedTransaction,
         log_context: &AdapterLogSchema,
-    ) -> (VMStatus, TransactionOutputExt) {
-        macro_rules! unwrap_or_discard {
-            ($res:expr) => {
-                match $res {
-                    Ok(s) => s,
-                    Err(e) => return discard_error_vm_status(e),
-                }
-            };
-        }
-
+        gas_meter: &mut G,
+    ) -> (VMStatus, TransactionOutputExt)
+    where
+        G: AptosGasMeter,
+        S: MoveResolverExt + StateView,
+    {
         // Revalidate the transaction.
         let resolver = self.0.new_move_resolver(storage);
         let mut session = self.0.new_session(&resolver, SessionId::txn(txn));
@@ -1011,15 +1052,8 @@ impl AptosVM {
             session = self.0.new_session(&resolver, SessionId::txn(txn));
         }
 
-        let gas_params = unwrap_or_discard!(self.0.get_gas_parameters(log_context));
         let storage_gas_params = unwrap_or_discard!(self.0.get_storage_gas_parameters(log_context));
         let txn_data = TransactionMetadata::new(txn);
-        let mut gas_meter = AptosGasMeter::new(
-            self.0.get_gas_feature_version(),
-            gas_params.clone(),
-            storage_gas_params.clone(),
-            txn_data.max_gas_amount(),
-        );
 
         // We keep track of whether any newly published modules are loaded into the Vm's loader
         // cache as part of executing transactions. This would allow us to decide whether the cache
@@ -1031,31 +1065,34 @@ impl AptosVM {
                 .execute_script_or_entry_function(
                     storage,
                     session,
-                    &mut gas_meter,
+                    gas_meter,
                     &txn_data,
                     payload,
                     log_context,
                     &mut new_published_modules_loaded,
+                    &storage_gas_params.change_set_configs,
                 ),
             TransactionPayload::Multisig(payload) => self.execute_multisig_transaction(
                 storage,
                 session,
-                &mut gas_meter,
+                gas_meter,
                 &txn_data,
                 payload,
                 log_context,
                 &mut new_published_modules_loaded,
+                &storage_gas_params.change_set_configs,
             ),
 
             // Deprecated. Will be removed in the future.
             TransactionPayload::ModuleBundle(m) => self.execute_modules(
                 storage,
                 session,
-                &mut gas_meter,
+                gas_meter,
                 &txn_data,
                 m,
                 log_context,
                 &mut new_published_modules_loaded,
+                &storage_gas_params.change_set_configs,
             ),
         };
 
@@ -1083,16 +1120,62 @@ impl AptosVM {
                 } else {
                     self.failed_transaction_cleanup_and_keep_vm_status(
                         err,
-                        &mut gas_meter,
+                        gas_meter,
                         &txn_data,
                         storage,
                         log_context,
+                        &storage_gas_params.change_set_configs,
                     )
                 }
             },
         }
     }
 
+    pub(crate) fn execute_user_transaction<S: MoveResolverExt + StateView>(
+        &self,
+        storage: &S,
+        txn: &SignatureCheckedTransaction,
+        log_context: &AdapterLogSchema,
+    ) -> (VMStatus, TransactionOutputExt) {
+        let balance = TransactionMetadata::new(txn).max_gas_amount();
+        // TODO: would we end up having a diverging behavior by creating the gas meter at an earlier time?
+        let mut gas_meter = unwrap_or_discard!(self.make_standard_gas_meter(balance, log_context));
+
+        self.execute_user_transaction_impl(storage, txn, log_context, &mut gas_meter)
+    }
+
+    pub fn execute_user_transaction_with_custom_gas_meter<S, G, F>(
+        state_view: &S,
+        txn: &SignatureCheckedTransaction,
+        log_context: &AdapterLogSchema,
+        make_gas_meter: F,
+    ) -> Result<(VMStatus, TransactionOutput, G), VMStatus>
+    where
+        S: StateView,
+        G: AptosGasMeter,
+        F: FnOnce(u64, AptosGasParameters, StorageGasParameters, Gas) -> Result<G, VMStatus>,
+    {
+        // TODO(Gas): revisit this.
+        init_speculative_logs(1);
+
+        let storage = StorageAdapter::new(state_view);
+        let vm = AptosVM::new(&storage);
+
+        // TODO(Gas): avoid creating txn metadata twice.
+        let balance = TransactionMetadata::new(txn).max_gas_amount();
+        let mut gas_meter = make_gas_meter(
+            vm.0.get_gas_feature_version(),
+            vm.0.get_gas_parameters(log_context)?.clone(),
+            vm.0.get_storage_gas_parameters(log_context)?.clone(),
+            balance,
+        )?;
+
+        let (status, output) =
+            vm.execute_user_transaction_impl(&storage, txn, log_context, &mut gas_meter);
+
+        Ok((status, output.into_transaction_output(&storage), gas_meter))
+    }
+
     fn execute_writeset<S: MoveResolverExt>(
         &self,
         storage: &S,
@@ -1288,7 +1371,7 @@ impl AptosVM {
     ) -> Result<Vec<Vec<u8>>> {
         let vm = AptosVM::new(state_view);
         let log_context = AdapterLogSchema::new(state_view.id(), 0);
-        let mut gas_meter = AptosGasMeter::new(
+        let mut gas_meter = StandardGasMeter::new(
             vm.0.get_gas_feature_version(),
             vm.0.get_gas_parameters(&log_context)?.clone(),
             vm.0.get_storage_gas_parameters(&log_context)?.clone(),
@@ -1658,7 +1741,7 @@ impl AptosSimulationVM {
             Ok(s) => s,
         };
 
-        let mut gas_meter = AptosGasMeter::new(
+        let mut gas_meter = StandardGasMeter::new(
             self.0 .0.get_gas_feature_version(),
             gas_params.clone(),
             storage_gas_params.clone(),
@@ -1677,6 +1760,7 @@ impl AptosSimulationVM {
                     payload,
                     log_context,
                     &mut new_published_modules_loaded,
+                    &storage_gas_params.change_set_configs,
                 )
             },
             TransactionPayload::Multisig(multisig) => {
@@ -1697,12 +1781,12 @@ impl AptosSimulationVM {
                                     // which is in the middle. Introducing a boolean would make the code
                                     // messier.
                                     let change_set_ext = session
-                                        .finish(&mut (), gas_meter.change_set_configs())
+                                        .finish(&mut (), &storage_gas_params.change_set_configs)
                                         .map_err(|e| e.into_vm_status())?;
-                                    gas_meter.charge_write_set_gas_for_io(
+                                    gas_meter.charge_io_gas_for_write_set(
                                         change_set_ext.write_set().iter(),
                                     )?;
-                                    gas_meter.charge_storage_fee(
+                                    gas_meter.charge_storage_fee_for_all(
                                         change_set_ext.write_set().iter(),
                                         change_set_ext.change_set().events(),
                                         txn_data.transaction_size,
@@ -1714,6 +1798,7 @@ impl AptosSimulationVM {
                                         &mut gas_meter,
                                         &txn_data,
                                         log_context,
+                                        &storage_gas_params.change_set_configs,
                                     )
                                 })
                         },
@@ -1732,6 +1817,7 @@ impl AptosSimulationVM {
                 m,
                 log_context,
                 &mut new_published_modules_loaded,
+                &storage_gas_params.change_set_configs,
             ),
         };
 
@@ -1756,6 +1842,7 @@ impl AptosSimulationVM {
                         &txn_data,
                         storage,
                         log_context,
+                        &storage_gas_params.change_set_configs,
                     );
                     (vm_status, output)
                 }
diff --git a/aptos-move/e2e-testsuite/src/tests/failed_transaction_tests.rs b/aptos-move/e2e-testsuite/src/tests/failed_transaction_tests.rs
index 45be28582f2b2..42b7cb370de70 100644
--- a/aptos-move/e2e-testsuite/src/tests/failed_transaction_tests.rs
+++ b/aptos-move/e2e-testsuite/src/tests/failed_transaction_tests.rs
@@ -3,7 +3,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use aptos_gas::{
-    AptosGasMeter, AptosGasParameters, StorageGasParameters, LATEST_GAS_FEATURE_VERSION,
+    AptosGasParameters, StandardGasMeter, StorageGasParameters, LATEST_GAS_FEATURE_VERSION,
 };
 use aptos_language_e2e_tests::{common_transactions::peer_to_peer_txn, executor::FakeExecutor};
 use aptos_state_view::TStateView;
@@ -36,7 +36,10 @@ fn failed_transaction_cleanup_test() {
 
     let gas_params = AptosGasParameters::zeros();
     let storage_gas_params = StorageGasParameters::free_and_unlimited();
-    let mut gas_meter = AptosGasMeter::new(
+
+    let change_set_configs = storage_gas_params.change_set_configs.clone();
+
+    let mut gas_meter = StandardGasMeter::new(
         LATEST_GAS_FEATURE_VERSION,
         gas_params,
         storage_gas_params,
@@ -50,6 +53,7 @@ fn failed_transaction_cleanup_test() {
         &txn_data,
         &data_cache,
         &log_context,
+        &change_set_configs,
     );
     assert!(!out1.txn_output().write_set().is_empty());
     assert_eq!(out1.txn_output().gas_used(), 90_000);
@@ -67,6 +71,7 @@ fn failed_transaction_cleanup_test() {
         &txn_data,
         &data_cache,
         &log_context,
+        &change_set_configs,
     );
     assert!(out2.txn_output().write_set().is_empty());
     assert!(out2.txn_output().gas_used() == 0);
diff --git a/crates/aptos/Cargo.toml b/crates/aptos/Cargo.toml
index ff7b6c1c4e1c3..636d536e1b17f 100644
--- a/crates/aptos/Cargo.toml
+++ b/crates/aptos/Cargo.toml
@@ -21,9 +21,11 @@ aptos-cached-packages = { workspace = true }
 aptos-config = { workspace = true }
 aptos-crypto = { workspace = true }
 aptos-db-tool = { workspace = true }
+aptos-debugger = { workspace = true }
 aptos-faucet-core = { workspace = true }
 aptos-framework = { workspace = true }
 aptos-gas = { workspace = true }
+aptos-gas-profiling = { workspace = true }
 aptos-genesis = { workspace = true }
 aptos-github-client = { workspace = true }
 aptos-global-constants = { workspace = true }
diff --git a/crates/aptos/src/common/types.rs b/crates/aptos/src/common/types.rs
index 4857c72d0bf25..978eb5ca1e928 100644
--- a/crates/aptos/src/common/types.rs
+++ b/crates/aptos/src/common/types.rs
@@ -18,6 +18,8 @@ use aptos_crypto::{
     ed25519::{Ed25519PrivateKey, Ed25519PublicKey, Ed25519Signature},
     x25519, PrivateKey, ValidCryptoMaterial, ValidCryptoMaterialStringExt,
 };
+use aptos_debugger::AptosDebugger;
+use aptos_gas_profiling::FrameName;
 use aptos_global_constants::adjust_gas_headroom;
 use aptos_keygen::KeyGen;
 use aptos_rest_client::{
@@ -28,7 +30,9 @@ use aptos_rest_client::{
 use aptos_sdk::{transaction_builder::TransactionFactory, types::LocalAccount};
 use aptos_types::{
     chain_id::ChainId,
-    transaction::{authenticator::AuthenticationKey, SignedTransaction, TransactionPayload},
+    transaction::{
+        authenticator::AuthenticationKey, SignedTransaction, TransactionPayload, TransactionStatus,
+    },
 };
 use async_trait::async_trait;
 use clap::{ArgEnum, Parser};
@@ -1288,6 +1292,11 @@ pub struct TransactionOptions {
     pub(crate) gas_options: GasOptions,
     #[clap(flatten)]
     pub(crate) prompt_options: PromptOptions,
+
+    /// If this option is set, simulate the transaction locally using the debugger and generate
+    /// flamegraphs that reflect the gas usage.
+    #[clap(long)]
+    pub(crate) profile_gas: bool,
 }
 
 impl TransactionOptions {
@@ -1442,6 +1451,170 @@ impl TransactionOptions {
         Ok(response.into_inner())
     }
 
+    /// Simulate the transaction locally using the debugger, with the gas profiler enabled.
+    pub async fn profile_gas(
+        &self,
+        payload: TransactionPayload,
+    ) -> CliTypedResult<TransactionSummary> {
+        println!();
+        println!("Simulating transaction locally with the gas profiler...");
+        println!("This is still experimental so results may be inaccurate.");
+
+        let client = self.rest_client()?;
+
+        // Fetch the chain states required for the simulation
+        // TODO(Gas): get the following from the chain
+        const DEFAULT_GAS_UNIT_PRICE: u64 = 100;
+        const DEFAULT_MAX_GAS: u64 = 2_000_000;
+
+        let (sender_key, sender_address) = self.get_key_and_address()?;
+        let gas_unit_price = self
+            .gas_options
+            .gas_unit_price
+            .unwrap_or(DEFAULT_GAS_UNIT_PRICE);
+        let (account, state) = get_account_with_state(&client, sender_address).await?;
+        let version = state.version;
+        let chain_id = ChainId::new(state.chain_id);
+        let sequence_number = account.sequence_number;
+
+        let balance = client
+            .get_account_balance_at_version(sender_address, version)
+            .await
+            .map_err(|err| CliError::ApiError(err.to_string()))?
+            .into_inner();
+
+        let max_gas = self.gas_options.max_gas.unwrap_or_else(|| {
+            if gas_unit_price == 0 {
+                DEFAULT_MAX_GAS
+            } else {
+                std::cmp::min(balance.coin.value.0 / gas_unit_price, DEFAULT_MAX_GAS)
+            }
+        });
+
+        // Create and sign the transaction
+        let transaction_factory = TransactionFactory::new(chain_id)
+            .with_gas_unit_price(gas_unit_price)
+            .with_max_gas_amount(max_gas)
+            .with_transaction_expiration_time(self.gas_options.expiration_secs);
+        let sender_account = &mut LocalAccount::new(sender_address, sender_key, sequence_number);
+        let transaction =
+            sender_account.sign_with_transaction_builder(transaction_factory.payload(payload));
+        let hash = transaction.clone().committed_hash();
+
+        // Execute the transaction using the debugger
+        let debugger = AptosDebugger::rest_client(client).unwrap();
+        let res = debugger.execute_transaction_at_version_with_gas_profiler(version, transaction);
+        let (vm_status, output, gas_log) = res.map_err(|err| {
+            CliError::UnexpectedError(format!("failed to simulate txn with gas profiler: {}", err))
+        })?;
+
+        // Generate the file name for the flamegraphs
+        let entry_point = gas_log.entry_point();
+
+        let human_readable_name = match entry_point {
+            FrameName::Script => "script".to_string(),
+            FrameName::Function {
+                module_id, name, ..
+            } => {
+                let addr_short = module_id.address().short_str_lossless();
+                let addr_truncated = if addr_short.len() > 4 {
+                    &addr_short[..4]
+                } else {
+                    addr_short.as_str()
+                };
+                format!("0x{}-{}-{}", addr_truncated, module_id.name(), name)
+            },
+        };
+        let raw_file_name = format!("txn-{}-{}", hash, human_readable_name);
+
+        // Create the directory if it does not exist yet.
+        let dir: &Path = Path::new("gas-profiling");
+
+        macro_rules! create_dir {
+            () => {
+                if let Err(err) = std::fs::create_dir(dir) {
+                    if err.kind() != std::io::ErrorKind::AlreadyExists {
+                        return Err(CliError::UnexpectedError(format!(
+                            "failed to create directory {}",
+                            dir.display()
+                        )));
+                    }
+                }
+            };
+        }
+
+        // Generate the execution & IO flamegraph.
+        println!();
+        match gas_log.to_flamegraph(format!("Transaction {} -- Execution & IO", hash))? {
+            Some(graph_bytes) => {
+                create_dir!();
+                let graph_file_path = Path::join(dir, format!("{}.exec_io.svg", raw_file_name));
+                std::fs::write(&graph_file_path, graph_bytes).map_err(|err| {
+                    CliError::UnexpectedError(format!(
+                        "Failed to write flamegraph to file {} : {:?}",
+                        graph_file_path.display(),
+                        err
+                    ))
+                })?;
+                println!(
+                    "Execution & IO Gas flamegraph saved to {}",
+                    graph_file_path.display()
+                );
+            },
+            None => {
+                println!("Skipped generating execution & IO flamegraph");
+            },
+        }
+
+        // Generate the storage fee flamegraph.
+        match gas_log
+            .storage
+            .to_flamegraph(format!("Transaction {} -- Storage Fee", hash))?
+        {
+            Some(graph_bytes) => {
+                create_dir!();
+                let graph_file_path = Path::join(dir, format!("{}.storage.svg", raw_file_name));
+                std::fs::write(&graph_file_path, graph_bytes).map_err(|err| {
+                    CliError::UnexpectedError(format!(
+                        "Failed to write flamegraph to file {} : {:?}",
+                        graph_file_path.display(),
+                        err
+                    ))
+                })?;
+                println!(
+                    "Storage fee flamegraph saved to {}",
+                    graph_file_path.display()
+                );
+            },
+            None => {
+                println!("Skipped generating storage fee flamegraph");
+            },
+        }
+
+        println!();
+
+        // Generate the transaction summary
+
+        // TODO(Gas): double check if this is correct.
+        let success = match output.status() {
+            TransactionStatus::Keep(exec_status) => Some(exec_status.is_success()),
+            TransactionStatus::Discard(_) | TransactionStatus::Retry => None,
+        };
+
+        Ok(TransactionSummary {
+            transaction_hash: hash.into(),
+            gas_used: Some(output.gas_used()),
+            gas_unit_price: Some(gas_unit_price),
+            pending: None,
+            sender: Some(sender_address),
+            sequence_number: None, // The transaction is not comitted so there is no new sequence number.
+            success,
+            timestamp_us: None,
+            version: Some(version), // The transaction is not comitted so there is no new version.
+            vm_status: Some(vm_status.to_string()),
+        })
+    }
+
     pub async fn estimate_gas_price(&self) -> CliTypedResult<u64> {
         let client = self.rest_client()?;
         client
diff --git a/crates/aptos/src/move_tool/mod.rs b/crates/aptos/src/move_tool/mod.rs
index c859c8d5867de..1a346ab26450b 100644
--- a/crates/aptos/src/move_tool/mod.rs
+++ b/crates/aptos/src/move_tool/mod.rs
@@ -744,10 +744,14 @@ impl CliCommand<TransactionSummary> for PublishPackage {
                 MAX_PUBLISH_PACKAGE_SIZE, size
             )));
         }
-        txn_options
-            .submit_transaction(payload)
-            .await
-            .map(TransactionSummary::from)
+        if txn_options.profile_gas {
+            txn_options.profile_gas(payload).await
+        } else {
+            txn_options
+                .submit_transaction(payload)
+                .await
+                .map(TransactionSummary::from)
+        }
     }
 }
 
@@ -1133,15 +1137,21 @@ impl CliCommand<TransactionSummary> for RunFunction {
             type_args.push(type_tag)
         }
 
-        self.txn_options
-            .submit_transaction(TransactionPayload::EntryFunction(EntryFunction::new(
-                self.function_id.module_id,
-                self.function_id.member_id,
-                type_args,
-                args,
-            )))
-            .await
-            .map(TransactionSummary::from)
+        let payload = TransactionPayload::EntryFunction(EntryFunction::new(
+            self.function_id.module_id,
+            self.function_id.member_id,
+            type_args,
+            args,
+        ));
+
+        if self.txn_options.profile_gas {
+            self.txn_options.profile_gas(payload).await
+        } else {
+            self.txn_options
+                .submit_transaction(payload)
+                .await
+                .map(TransactionSummary::from)
+        }
     }
 }
 
@@ -1243,13 +1253,16 @@ impl CliCommand<TransactionSummary> for RunScript {
             type_args.push(type_tag)
         }
 
-        let txn = self
-            .txn_options
-            .submit_transaction(TransactionPayload::Script(Script::new(
-                bytecode, type_args, args,
-            )))
-            .await?;
-        Ok(TransactionSummary::from(&txn))
+        let payload = TransactionPayload::Script(Script::new(bytecode, type_args, args));
+
+        if self.txn_options.profile_gas {
+            self.txn_options.profile_gas(payload).await
+        } else {
+            self.txn_options
+                .submit_transaction(payload)
+                .await
+                .map(TransactionSummary::from)
+        }
     }
 }