Migrates the SHA256 hasher out of the threadpool
This commit is contained in:
+3
-2
@@ -32,7 +32,7 @@ import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
|||||||
import com.vitorpamplona.quartz.nip01Core.core.fastForEach
|
import com.vitorpamplona.quartz.nip01Core.core.fastForEach
|
||||||
import com.vitorpamplona.quartz.nip01Core.jackson.JacksonMapper
|
import com.vitorpamplona.quartz.nip01Core.jackson.JacksonMapper
|
||||||
import com.vitorpamplona.quartz.utils.Hex
|
import com.vitorpamplona.quartz.utils.Hex
|
||||||
import com.vitorpamplona.quartz.utils.sha256.pool
|
import com.vitorpamplona.quartz.utils.sha256.threadLocalDigest
|
||||||
import java.io.IOException
|
import java.io.IOException
|
||||||
|
|
||||||
actual object EventHasherSerializer {
|
actual object EventHasherSerializer {
|
||||||
@@ -127,7 +127,8 @@ actual object EventHasherSerializer {
|
|||||||
content: String,
|
content: String,
|
||||||
): Boolean {
|
): Boolean {
|
||||||
val br: BufferRecycler = JacksonMapper.mapper.factory._getBufferRecycler()
|
val br: BufferRecycler = JacksonMapper.mapper.factory._getBufferRecycler()
|
||||||
val bb = HashingByteArrayBuilder(br, pool)
|
val digest = threadLocalDigest.get()
|
||||||
|
val bb = HashingByteArrayBuilder(br, digest)
|
||||||
try {
|
try {
|
||||||
val generator = JacksonMapper.mapper.createGenerator(bb, JsonEncoding.UTF8)
|
val generator = JacksonMapper.mapper.createGenerator(bb, JsonEncoding.UTF8)
|
||||||
generator.enable(JsonGenerator.Feature.COMBINE_UNICODE_SURROGATES_IN_UTF8)
|
generator.enable(JsonGenerator.Feature.COMBINE_UNICODE_SURROGATES_IN_UTF8)
|
||||||
|
|||||||
+3
-8
@@ -21,8 +21,8 @@
|
|||||||
package com.vitorpamplona.quartz.nip01Core.crypto
|
package com.vitorpamplona.quartz.nip01Core.crypto
|
||||||
|
|
||||||
import com.fasterxml.jackson.core.util.BufferRecycler
|
import com.fasterxml.jackson.core.util.BufferRecycler
|
||||||
import com.vitorpamplona.quartz.utils.sha256.Sha256Pool
|
|
||||||
import java.io.OutputStream
|
import java.io.OutputStream
|
||||||
|
import java.security.MessageDigest
|
||||||
import kotlin.math.max
|
import kotlin.math.max
|
||||||
import kotlin.math.min
|
import kotlin.math.min
|
||||||
|
|
||||||
@@ -34,20 +34,15 @@ import kotlin.math.min
|
|||||||
*/
|
*/
|
||||||
class HashingByteArrayBuilder(
|
class HashingByteArrayBuilder(
|
||||||
private val bufferRecycler: BufferRecycler,
|
private val bufferRecycler: BufferRecycler,
|
||||||
private val hashPool: Sha256Pool,
|
private val hasher: MessageDigest,
|
||||||
) : OutputStream() {
|
) : OutputStream() {
|
||||||
var buf: ByteArray = bufferRecycler.allocByteBuffer(BufferRecycler.BYTE_WRITE_CONCAT_BUFFER)
|
var buf: ByteArray = bufferRecycler.allocByteBuffer(BufferRecycler.BYTE_WRITE_CONCAT_BUFFER)
|
||||||
var bufLength: Int = 0
|
var bufLength: Int = 0
|
||||||
|
|
||||||
// lazy minimizes lock conflicts with the pool
|
|
||||||
val hasher by lazy {
|
|
||||||
hashPool.acquire()
|
|
||||||
}
|
|
||||||
|
|
||||||
fun release() {
|
fun release() {
|
||||||
bufLength = 0
|
bufLength = 0
|
||||||
bufferRecycler.releaseByteBuffer(BufferRecycler.BYTE_WRITE_CONCAT_BUFFER, buf)
|
bufferRecycler.releaseByteBuffer(BufferRecycler.BYTE_WRITE_CONCAT_BUFFER, buf)
|
||||||
hashPool.release(hasher)
|
hasher.reset()
|
||||||
buf = EMPTY_ARRAY
|
buf = EMPTY_ARRAY
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+11
-10
@@ -30,16 +30,13 @@ import java.security.MessageDigest
|
|||||||
* (lock acquire + release) for ~2µs of actual hashing. ThreadLocal eliminates all locking
|
* (lock acquire + release) for ~2µs of actual hashing. ThreadLocal eliminates all locking
|
||||||
* since each thread gets its own MessageDigest instance. digest() implicitly resets state.
|
* since each thread gets its own MessageDigest instance. digest() implicitly resets state.
|
||||||
*/
|
*/
|
||||||
private val threadLocalDigest =
|
val threadLocalDigest =
|
||||||
ThreadLocal.withInitial {
|
ThreadLocal.withInitial {
|
||||||
MessageDigest.getInstance("SHA-256")
|
MessageDigest.getInstance("SHA-256")
|
||||||
}
|
}
|
||||||
|
|
||||||
actual fun sha256(data: ByteArray): ByteArray = threadLocalDigest.get().digest(data)
|
actual fun sha256(data: ByteArray): ByteArray = threadLocalDigest.get().digest(data)
|
||||||
|
|
||||||
/** Pool for incremental hashing (used by EventHasherSerializer, HashingByteArrayBuilder). */
|
|
||||||
val pool = Sha256Pool(25)
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Calculate SHA256 hash while counting bytes read from the stream.
|
* Calculate SHA256 hash while counting bytes read from the stream.
|
||||||
* Returns both the hash and the number of bytes processed.
|
* Returns both the hash and the number of bytes processed.
|
||||||
@@ -55,11 +52,15 @@ fun sha256StreamWithCount(
|
|||||||
): Pair<ByteArray, Long> {
|
): Pair<ByteArray, Long> {
|
||||||
val countingStream = CountingInputStream(inputStream)
|
val countingStream = CountingInputStream(inputStream)
|
||||||
val digest = threadLocalDigest.get()
|
val digest = threadLocalDigest.get()
|
||||||
val buffer = ByteArray(bufferSize)
|
try {
|
||||||
var bytesRead: Int
|
val buffer = ByteArray(bufferSize)
|
||||||
while (countingStream.read(buffer).also { bytesRead = it } != -1) {
|
var bytesRead: Int
|
||||||
digest.update(buffer, 0, bytesRead)
|
while (countingStream.read(buffer).also { bytesRead = it } != -1) {
|
||||||
|
digest.update(buffer, 0, bytesRead)
|
||||||
|
}
|
||||||
|
return Pair(digest.digest(), countingStream.bytesRead)
|
||||||
|
} catch (e: Exception) {
|
||||||
|
digest.reset() // Clean up state on failure
|
||||||
|
throw e
|
||||||
}
|
}
|
||||||
val hash = digest.digest()
|
|
||||||
return Pair(hash, countingStream.bytesRead)
|
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user