{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/19222","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/19222/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/19222/comments","events_url":"https://api.github.com/repos/apache/spark/issues/19222/events","html_url":"https://github.com/apache/spark/pull/19222","id":257470094,"number":19222,"title":"[SPARK-10399][CORE][SQL] Introduce multiple MemoryBlocks to choose several types of memory block","user":{"login":"kiszk","id":1315079,"avatar_url":"https://avatars2.githubusercontent.com/u/1315079?v=4","gravatar_id":"","url":"https://api.github.com/users/kiszk","html_url":"https://github.com/kiszk","followers_url":"https://api.github.com/users/kiszk/followers","following_url":"https://api.github.com/users/kiszk/following{/other_user}","gists_url":"https://api.github.com/users/kiszk/gists{/gist_id}","starred_url":"https://api.github.com/users/kiszk/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/kiszk/subscriptions","organizations_url":"https://api.github.com/users/kiszk/orgs","repos_url":"https://api.github.com/users/kiszk/repos","events_url":"https://api.github.com/users/kiszk/events{/privacy}","received_events_url":"https://api.github.com/users/kiszk/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":119,"created_at":"2017-09-13T18:05:08Z","updated_at":"2018-03-29T06:38:02Z","closed_at":null,"author_association":"CONTRIBUTOR","pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/19222","html_url":"https://github.com/apache/spark/pull/19222","diff_url":"https://github.com/apache/spark/pull/19222.diff","patch_url":"https://github.com/apache/spark/pull/19222.patch"},"body":"## What changes were proposed in this pull request?\r\n\r\nThis PR allows us to use one of several types of `MemoryBlock`, such as byte array, int array, long array, or `java.nio.DirectByteBuffer`. To use `java.nio.DirectByteBuffer` allows to have off heap memory which is automatically deallocated by JVM. `MemoryBlock`  class has primitive accessors like `Platform.getInt()`, `Platform.putint()`, or `Platform.copyMemory()`. \r\n\r\nThis PR uses `MemoryBlock` for `OffHeapColumnVector`, `UTF8String`, and other places. This PR can improve performance of operations involving memory accesses (e.g. `UTF8String.trim`) by 1.8x.\r\n\r\nFor now, this PR does not use `MemoryBlock` for `BufferHolder` based on @cloud-fan's [suggestion](https://github.com/apache/spark/pull/11494#issuecomment-309694290).\r\n\r\nSince this PR is a successor of #11494, close #11494. Many codes were ported from #11494. Many efforts were put here. **I think this PR should credit to @yzotov.**\r\n\r\n\r\nThis PR can achieve **1.1-1.4x performance improvements** for  operations in `UTF8String` or `Murmur3_x86_32`. Other operations are almost comparable performances.\r\n\r\nWithout this PR\r\n```\r\nOpenJDK 64-Bit Server VM 1.8.0_121-8u121-b13-0ubuntu1.16.04.2-b13 on Linux 4.4.0-22-generic\r\nIntel(R) Xeon(R) CPU E5-2667 v3 @ 3.20GHz\r\nOpenJDK 64-Bit Server VM 1.8.0_121-8u121-b13-0ubuntu1.16.04.2-b13 on Linux 4.4.0-22-generic\r\nIntel(R) Xeon(R) CPU E5-2667 v3 @ 3.20GHz\r\nHash byte arrays with length 268435487:  Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n------------------------------------------------------------------------------------------------\r\nMurmur3_x86_32                                 526 /  536          0.0   131399881.5       1.0X\r\n\r\nUTF8String benchmark:                    Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n------------------------------------------------------------------------------------------------\r\nhashCode                                       525 /  552       1022.6           1.0       1.0X\r\nsubstring                                      414 /  423       1298.0           0.8       1.3X\r\n```\r\n\r\nWith this PR\r\n```\r\nOpenJDK 64-Bit Server VM 1.8.0_121-8u121-b13-0ubuntu1.16.04.2-b13 on Linux 4.4.0-22-generic\r\nIntel(R) Xeon(R) CPU E5-2667 v3 @ 3.20GHz\r\nHash byte arrays with length 268435487:  Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n------------------------------------------------------------------------------------------------\r\nMurmur3_x86_32                                 474 /  488          0.0   118552232.0       1.0X\r\n\r\nUTF8String benchmark:                    Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n------------------------------------------------------------------------------------------------\r\nhashCode                                       476 /  480       1127.3           0.9       1.0X\r\nsubstring                                      287 /  291       1869.9           0.5       1.7X\r\n```\r\n\r\nBenchmark program\r\n```\r\ntest(\"benchmark Murmur3_x86_32\") {\r\n  val length = 8192 * 32768 + 31\r\n  val seed = 42L\r\n  val iters = 1 << 2\r\n  val random = new Random(seed)\r\n  val arrays = Array.fill[MemoryBlock](numArrays) {\r\n    val bytes = new Array[Byte](length)\r\n    random.nextBytes(bytes)\r\n    new ByteArrayMemoryBlock(bytes, Platform.BYTE_ARRAY_OFFSET, length)\r\n  }\r\n                                                                \r\n  val benchmark = new Benchmark(\"Hash byte arrays with length \" + length,\r\n    iters * numArrays, minNumIters = 20) \r\n  benchmark.addCase(\"HiveHasher\") { _: Int =>\r\n    var sum = 0L\r\n    for (_ <- 0L until iters) {\r\n      sum += HiveHasher.hashUnsafeBytesBlock(\r\n        arrays(i), Platform.BYTE_ARRAY_OFFSET, length)\r\n    }\r\n  }\r\n  benchmark.run()   \r\n}\r\n\r\ntest(\"benchmark UTF8String\") {\r\n  val N = 512 * 1024 * 1024\r\n  val iters = 2\r\n  val benchmark = new Benchmark(\"UTF8String benchmark\", N, minNumIters = 20)\r\n  val str0 = new java.io.StringWriter() { { for (i <- 0 until N) { write(\" \") } } }.toString\r\n  val s0 = UTF8String.fromString(str0)\r\n  benchmark.addCase(\"hashCode\") { _: Int =>\r\n    var h: Int = 0\r\n    for (_ <- 0L until iters) { h += s0.hashCode }\r\n  }\r\n  benchmark.addCase(\"substring\") { _: Int =>\r\n    var s: UTF8String = null\r\n    for (_ <- 0L until iters) { s = s0.substring(N / 2 - 5, N / 2 + 5) }\r\n  }\r\n  benchmark.run()\r\n}\r\n```\r\n\r\n  \r\n  \r\nI run [this benchmark program](https://gist.github.com/kiszk/94f75b506c93a663bbbc372ffe8f05de) using [the commit](https://github.com/apache/spark/pull/19222/commits/ee5a79861c18725fb1cd9b518cdfd2489c05b81d6). I got the following results:\r\n\r\n```\r\nOpenJDK 64-Bit Server VM 1.8.0_151-8u151-b12-0ubuntu0.16.04.2-b12 on Linux 4.4.0-66-generic\r\nIntel(R) Xeon(R) CPU E5-2667 v3 @ 3.20GHz\r\nMemory access benchmarks:                Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n------------------------------------------------------------------------------------------------\r\nByteArrayMemoryBlock get/putInt()              220 /  221        609.3           1.6       1.0X\r\nPlatform get/putInt(byte[])                    220 /  236        610.9           1.6       1.0X\r\nPlatform get/putInt(Object)                    492 /  494        272.8           3.7       0.4X\r\nOnHeapMemoryBlock get/putLong()                322 /  323        416.5           2.4       0.7X\r\nlong[]                                         221 /  221        608.0           1.6       1.0X\r\nPlatform get/putLong(long[])                   321 /  321        418.7           2.4       0.7X\r\nPlatform get/putLong(Object)                   561 /  563        239.2           4.2       0.4X\r\n```\r\n\r\nI also run [this benchmark program](https://gist.github.com/kiszk/5fdb4e03733a5d110421177e289d1fb5) for comparing performance of `Platform.copyMemory()`.\r\n```\r\nOpenJDK 64-Bit Server VM 1.8.0_151-8u151-b12-0ubuntu0.16.04.2-b12 on Linux 4.4.0-66-generic\r\nIntel(R) Xeon(R) CPU E5-2667 v3 @ 3.20GHz\r\nPlatform copyMemory:                     Best/Avg Time(ms)    Rate(M/s)   Per Row(ns)   Relative\r\n------------------------------------------------------------------------------------------------\r\nObject to Object                              1961 / 1967          8.6         116.9       1.0X\r\nSystem.arraycopy Object to Object             1917 / 1921          8.8         114.3       1.0X\r\nbyte array to byte array                      1961 / 1968          8.6         116.9       1.0X\r\nSystem.arraycopy byte array to byte array      1909 / 1937          8.8         113.8       1.0X\r\nint array to int array                        1921 / 1990          8.7         114.5       1.0X\r\ndouble array to double array                  1918 / 1923          8.7         114.3       1.0X\r\nObject to byte array                          1961 / 1967          8.6         116.9       1.0X\r\nObject to short array                         1965 / 1972          8.5         117.1       1.0X\r\nObject to int array                           1910 / 1915          8.8         113.9       1.0X\r\nObject to float array                         1971 / 1978          8.5         117.5       1.0X\r\nObject to double array                        1919 / 1944          8.7         114.4       1.0X\r\nbyte array to Object                          1959 / 1967          8.6         116.8       1.0X\r\nint array to Object                           1961 / 1970          8.6         116.9       1.0X\r\ndouble array to Object                        1917 / 1924          8.8         114.3       1.0X\r\n```\r\n\r\nThese results show three facts:\r\n1. According to the second/third or sixth/seventh results in the first experiment, if we use `Platform.get/putInt(Object)`, we achieve more than 2x worse performance than `Platform.get/putInt(byte[])` with concrete type (i.e. `byte[]`).\r\n2. According to the second/third or fourth/fifth/sixth results in the first experiment, the fastest way to access an array element on Java heap is `array[]`. **Cons of `array[]` is that it is not possible to support unaligned-8byte access.**\r\n3. According to the first/second/third or fourth/sixth/seventh results in the first experiment, `getInt()/putInt() or getLong()/putLong()` in subclasses of `MemoryBlock` can achieve comparable performance to `Platform.get/putInt()` or `Platform.get/putLong()` with concrete type (second or sixth result). There is no overhead regarding virtual call.\r\n4. According to results in the second experiment, for `Platform.copy()`, to pass `Object` can achieve the same performance as to pass any type of primitive array as source or destination.\r\n5. According to second/fourth results in the second experiment, `Platform.copy()` can achieve the same performance as `System.arrayCopy`. **It would be good to use `Platform.copy()` since `Platform.copy()` can take any types for src and dst.**\r\n\r\nWe are incrementally replace `Platform.get/putXXX` with `MemoryBlock.get/putXXX`. This is because we have two advantages.\r\n1) Achieve better performance due to having a concrete type for an array.\r\n2) Use simple OO design instead of passing `Object`\r\nIt is easy to use `MemoryBlock` in `InternalRow`, `BufferHolder`, `TaskMemoryManager`, and others that are already abstracted. It is not easy to use `MemoryBlock` in utility classes related to hashing or others.\r\n\r\nOther candidates are \r\n- UnsafeRow, UnsafeArrayData, UnsafeMapData, SpecificUnsafeRowJoiner\r\n- UTF8StringBuffer\r\n- BufferHolder\r\n- TaskMemoryManager\r\n- OnHeapColumnVector\r\n- BytesToBytesMap\r\n- CachedBatch\r\n- classes for hash\r\n- others.\r\n\r\n\r\n## How was this patch tested?\r\n\r\nAdded `UnsafeMemoryAllocator`\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/377137337","html_url":"https://github.com/apache/spark/pull/19222#issuecomment-377137337","issue_url":"https://api.github.com/repos/apache/spark/issues/19222","id":377137337,"user":{"login":"SparkQA","id":8108735,"avatar_url":"https://avatars1.githubusercontent.com/u/8108735?v=4","gravatar_id":"","url":"https://api.github.com/users/SparkQA","html_url":"https://github.com/SparkQA","followers_url":"https://api.github.com/users/SparkQA/followers","following_url":"https://api.github.com/users/SparkQA/following{/other_user}","gists_url":"https://api.github.com/users/SparkQA/gists{/gist_id}","starred_url":"https://api.github.com/users/SparkQA/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/SparkQA/subscriptions","organizations_url":"https://api.github.com/users/SparkQA/orgs","repos_url":"https://api.github.com/users/SparkQA/repos","events_url":"https://api.github.com/users/SparkQA/events{/privacy}","received_events_url":"https://api.github.com/users/SparkQA/received_events","type":"User","site_admin":false},"created_at":"2018-03-29T06:38:02Z","updated_at":"2018-03-29T06:38:02Z","author_association":"NONE","body":"**[Test build #88693 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/88693/testReport)** for PR 19222 at commit [`b69cb64`](https://github.com/apache/spark/commit/b69cb6430d71fe6ce7a39f9d6a13bdcfa8704ccf)."}}
{"actor":{"display_login":"SparkQA"}}