mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-10-05 19:28:25 +00:00
Merge pull request #3533 from vitorpamplona/claude/amethyst-miner-performance-0rw1dn
Parallelize PoW mining with multi-worker nonce search
This commit is contained in:
@@ -29,6 +29,7 @@ import com.vitorpamplona.amethyst.commons.model.NoteState
|
||||
import com.vitorpamplona.amethyst.commons.relayClient.BlockedRelayFilteringClient
|
||||
import com.vitorpamplona.amethyst.commons.robohash.CachedRobohash
|
||||
import com.vitorpamplona.amethyst.commons.service.lnurl.OkHttpLnurlEndpointResolver
|
||||
import com.vitorpamplona.amethyst.commons.service.pow.PoWPolicy
|
||||
import com.vitorpamplona.amethyst.commons.service.pow.PoWPublishQueue
|
||||
import com.vitorpamplona.amethyst.commons.tor.TorSettings
|
||||
import com.vitorpamplona.amethyst.model.Account
|
||||
@@ -618,9 +619,12 @@ class AppModules(
|
||||
)
|
||||
|
||||
// fire-and-forget NIP-13 mining: posts queue here and publish when mined.
|
||||
// Capped worker pool so a burst of sends never spawns unbounded miners.
|
||||
// Template jobs checkpoint to disk (restored on login) and every enqueue
|
||||
// raises the shortService shield so backgrounding doesn't freeze a miner.
|
||||
// One job mines at a time, racing half the cores over disjoint nonce
|
||||
// slices — same total CPU budget as the old 2-job pool, but each post
|
||||
// finishes ~minerThreads× sooner and the other half of the cores stays
|
||||
// free for the UI. Template jobs checkpoint to disk (restored on login)
|
||||
// and every enqueue raises the shortService shield so backgrounding
|
||||
// doesn't freeze a miner.
|
||||
val powJobStore by lazy {
|
||||
PowJobStore(File(appContext.filesDir, PowJobStore.FILE_NAME), applicationIOScope)
|
||||
}
|
||||
@@ -628,7 +632,8 @@ class AppModules(
|
||||
val powPublishQueue by lazy {
|
||||
PoWPublishQueue(
|
||||
scope = applicationIOScope,
|
||||
maxConcurrent = (Runtime.getRuntime().availableProcessors() / 2).coerceIn(1, 2),
|
||||
maxConcurrent = 1,
|
||||
minerThreads = PoWPolicy.minerWorkers(Runtime.getRuntime().availableProcessors()),
|
||||
persistence = powJobStore,
|
||||
onQueueActive = { PowMiningForegroundService.start(appContext) },
|
||||
)
|
||||
|
||||
@@ -824,6 +824,14 @@ class Account(
|
||||
else -> overrideDifficulty
|
||||
}
|
||||
|
||||
/**
|
||||
* Parallel workers a single nonce search should use — the mining queue's
|
||||
* per-job budget (half the device's cores). 1 when no queue is wired.
|
||||
* Callers that run [PoWMiner] inside a queued job must pass this so the
|
||||
* job stays inside the queue's CPU budget.
|
||||
*/
|
||||
fun powMinerWorkers(): Int = powQueue()?.minerThreads ?: 1
|
||||
|
||||
/**
|
||||
* Enqueues [work] into the fire-and-forget mining queue. Returns false when
|
||||
* no queue is wired (headless/test accounts): callers must then run their
|
||||
@@ -932,7 +940,9 @@ class Account(
|
||||
mine = { isActive ->
|
||||
// the wrap's ephemeral key is generated inside the wrap build;
|
||||
// the conversion hook hands its pubkey back so the nonce can
|
||||
// commit to it.
|
||||
// commit to it. Single-threaded on purpose: the conversion is
|
||||
// a non-suspend hook deep inside the synchronous NIP-59 wrap
|
||||
// build, so it can't race PoWMiner.mine workers.
|
||||
val mineWrap: GiftWrapTemplateConversion = { template, ephemeralPubKey ->
|
||||
PoWMiner.run(template, ephemeralPubKey, difficulty, isActive)
|
||||
}
|
||||
@@ -964,14 +974,15 @@ class Account(
|
||||
isActive: () -> Boolean,
|
||||
): NostrSigner {
|
||||
val currentSigner = signer
|
||||
val workers = powMinerWorkers()
|
||||
return if (currentSigner is NostrSignerWithClientTag) {
|
||||
NostrSignerWithClientTag(
|
||||
inner = PoWNostrSigner(currentSigner.inner, difficulty, kindsToMine, isActive),
|
||||
inner = PoWNostrSigner(currentSigner.inner, difficulty, kindsToMine, isActive, workers),
|
||||
clientTag = currentSigner.clientTag,
|
||||
disabled = currentSigner.disabled,
|
||||
)
|
||||
} else {
|
||||
PoWNostrSigner(currentSigner, difficulty, kindsToMine, isActive)
|
||||
PoWNostrSigner(currentSigner, difficulty, kindsToMine, isActive, workers)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -22,10 +22,21 @@ package com.vitorpamplona.amethyst.service.pow
|
||||
|
||||
import android.content.Context
|
||||
import com.vitorpamplona.amethyst.R
|
||||
import com.vitorpamplona.amethyst.commons.service.pow.PoWEstimator
|
||||
import com.vitorpamplona.amethyst.commons.service.pow.PoWPolicy
|
||||
import com.vitorpamplona.amethyst.ui.pluralStringRes
|
||||
import com.vitorpamplona.amethyst.ui.stringRes
|
||||
import kotlin.math.roundToLong
|
||||
|
||||
/**
|
||||
* This device's effective mining rate: the [PoWEstimator] benchmark run with
|
||||
* the same worker count the mining queue uses (half the cores, see
|
||||
* [PoWPolicy.minerWorkers] and AppModules.powPublishQueue). Every UI estimate
|
||||
* must use this rate, or it would describe a single-threaded miner that no
|
||||
* longer exists. Cached after the first call (~250 ms).
|
||||
*/
|
||||
suspend fun deviceHashesPerSecond(): Double = PoWEstimator.hashesPerSecond(PoWPolicy.minerWorkers(Runtime.getRuntime().availableProcessors()))
|
||||
|
||||
/**
|
||||
* "45 seconds" / "10 minutes" / "3 hours" — the one human-readable rendering
|
||||
* of a PoW duration estimate, shared by the settings picker, the composer
|
||||
|
||||
@@ -73,6 +73,7 @@ import com.vitorpamplona.amethyst.commons.service.broadcast.BroadcastStatus
|
||||
import com.vitorpamplona.amethyst.commons.service.broadcast.RelayResult
|
||||
import com.vitorpamplona.amethyst.commons.service.pow.PoWEstimator
|
||||
import com.vitorpamplona.amethyst.commons.service.pow.PoWJobState
|
||||
import com.vitorpamplona.amethyst.service.pow.deviceHashesPerSecond
|
||||
import com.vitorpamplona.amethyst.service.pow.formatTimeLeft
|
||||
import com.vitorpamplona.amethyst.service.pow.powKindLabelRes
|
||||
import com.vitorpamplona.amethyst.ui.stringRes
|
||||
@@ -197,7 +198,7 @@ private fun MiningContent(
|
||||
val context = LocalContext.current
|
||||
val hashRate by
|
||||
produceState<Double?>(initialValue = null) {
|
||||
value = PoWEstimator.hashesPerSecond()
|
||||
value = deviceHashesPerSecond()
|
||||
}
|
||||
|
||||
Column(modifier = Modifier.fillMaxWidth()) {
|
||||
|
||||
+2
-1
@@ -48,6 +48,7 @@ import com.vitorpamplona.amethyst.R
|
||||
import com.vitorpamplona.amethyst.commons.icons.symbols.Icon
|
||||
import com.vitorpamplona.amethyst.commons.icons.symbols.MaterialSymbols
|
||||
import com.vitorpamplona.amethyst.commons.service.pow.PoWEstimator
|
||||
import com.vitorpamplona.amethyst.service.pow.deviceHashesPerSecond
|
||||
import com.vitorpamplona.amethyst.service.pow.formatApproxDuration
|
||||
import com.vitorpamplona.amethyst.ui.stringRes
|
||||
import com.vitorpamplona.amethyst.ui.theme.ThemeComparisonRow
|
||||
@@ -117,7 +118,7 @@ fun PowOverrideButton(
|
||||
val context = LocalContext.current
|
||||
val hashRate by
|
||||
produceState<Double?>(initialValue = null) {
|
||||
value = PoWEstimator.hashesPerSecond()
|
||||
value = deviceHashesPerSecond()
|
||||
}
|
||||
|
||||
fun eta(difficulty: Int): String? = hashRate?.let { formatApproxDuration(context, PoWEstimator.estimateSeconds(difficulty, it)) }
|
||||
|
||||
+1
-1
@@ -575,7 +575,7 @@ open class CommentPostViewModel :
|
||||
// fresh created_at at mining start (NIP-13 recommendation):
|
||||
// the job may have waited in the queue behind other posts.
|
||||
val fresh = EventTemplate<Event>(TimeUtils.now(), template.kind, template.tags, template.content)
|
||||
val mined = PoWMiner.run(fresh, anonSigner.pubKey, powDifficulty, isActive)
|
||||
val mined = PoWMiner.mine(fresh, anonSigner.pubKey, powDifficulty, accountViewModel.account.powMinerWorkers(), isActive)
|
||||
accountViewModel.account.signAnonymouslyAndBroadcast(mined, extraNotesToBroadcast, anonSigner)
|
||||
accountViewModel.account.deleteDraftIgnoreErrors(draftToDelete)
|
||||
}
|
||||
|
||||
+1
-1
@@ -1056,7 +1056,7 @@ open class ShortNotePostViewModel :
|
||||
// fresh created_at at mining start (NIP-13 recommendation):
|
||||
// the job may have waited in the queue behind other posts.
|
||||
val fresh = EventTemplate<Event>(TimeUtils.now(), template.kind, template.tags, template.content)
|
||||
val mined = PoWMiner.run(fresh, anonSigner.pubKey, powDifficulty, isActive)
|
||||
val mined = PoWMiner.mine(fresh, anonSigner.pubKey, powDifficulty, accountViewModel.account.powMinerWorkers(), isActive)
|
||||
accountViewModel.account.signAnonymouslyAndBroadcast(mined, extraNotesToBroadcast, anonSigner)
|
||||
accountViewModel.account.deleteDraftIgnoreErrors(draftToDelete)
|
||||
}
|
||||
|
||||
+2
-1
@@ -52,6 +52,7 @@ import com.vitorpamplona.amethyst.commons.service.pow.PoWEstimator
|
||||
import com.vitorpamplona.amethyst.model.AccountPoWPreferences
|
||||
import com.vitorpamplona.amethyst.model.BooleanType
|
||||
import com.vitorpamplona.amethyst.model.UiSettingsFlow
|
||||
import com.vitorpamplona.amethyst.service.pow.deviceHashesPerSecond
|
||||
import com.vitorpamplona.amethyst.service.pow.formatApproxDuration
|
||||
import com.vitorpamplona.amethyst.ui.navigation.navs.INav
|
||||
import com.vitorpamplona.amethyst.ui.navigation.topbars.TopBarWithBackButton
|
||||
@@ -195,7 +196,7 @@ private fun PowTimeEstimate(difficulty: Int) {
|
||||
val context = LocalContext.current
|
||||
val estimate by
|
||||
produceState<String?>(initialValue = null, difficulty) {
|
||||
val rate = PoWEstimator.hashesPerSecond()
|
||||
val rate = deviceHashesPerSecond()
|
||||
value = formatApproxDuration(context, PoWEstimator.estimateSeconds(difficulty, rate))
|
||||
}
|
||||
|
||||
|
||||
+3
-3
@@ -216,8 +216,8 @@ Army-knife verbs that operate purely on their arguments. They never touch
|
||||
| `amy encode naddr --kind N --pubkey HEX --identifier D [--relay URL[,URL…]]` | Encode an addressable-event (`a` tag) pointer. |
|
||||
| `amy verify [EVENT-JSON]` | Check an event's id hash and signature. Reads stdin when the argument is omitted or `-`. Reports `id_ok` + `signature_ok` separately. |
|
||||
| `amy pow check EVENT-JSON\|-` | NIP-13 difficulty of a signed event: `actual_bits`, `committed_target`, `has_commitment`, and `effective_pow` (capped at the commitment so lucky low-target spam doesn't over-count), plus `valid` (id + signature). |
|
||||
| `amy pow mine --target N [--pubkey HEX] [--timeout SECS] TEMPLATE-JSON\|-` | Mine an **unsigned** template to N leading zero bits and print it back with the nonce tag. Ids don't commit to signatures, so amy can mine on behalf of any pubkey (NIP-13 delegated PoW); defaults to the active account. Exit 124 on timeout. |
|
||||
| `amy pow bench` | Benchmark this machine's hash rate and print expected mining time at 16/20/24/28 bits. |
|
||||
| `amy pow mine --target N [--pubkey HEX] [--timeout SECS] [--threads N] TEMPLATE-JSON\|-` | Mine an **unsigned** template to N leading zero bits and print it back with the nonce tag. Ids don't commit to signatures, so amy can mine on behalf of any pubkey (NIP-13 delegated PoW); defaults to the active account. Mines on all cores by default (`--threads` to override). Exit 124 on timeout. |
|
||||
| `amy pow bench` | Benchmark this machine's hash rate — `hashes_per_second` is the all-cores rate `pow mine` uses by default, `hashes_per_second_single_core` the one-thread rate — and print expected mining time at 16/20/24/28 bits. |
|
||||
| `amy key generate` | Mint a fresh keypair (`nsec` + `npub` + hex). Does not persist — use `init`/`login` for that. |
|
||||
| `amy key public NSEC\|HEX` | Derive the public key from a secret key. |
|
||||
| `amy key encrypt NSEC\|HEX --password X` | NIP-49 encrypt a secret key to an `ncryptsec1…`. |
|
||||
@@ -384,7 +384,7 @@ HTTP endpoint. Reuses quartz's `Nip86Client` and the shared `Nip86Retriever`
|
||||
|
||||
| Command | What it does |
|
||||
|---|---|
|
||||
| `amy notes post TEXT [--relay URL] [--pow BITS [--pow-timeout SECS]]` | Publish a kind:1 short text note; `--pow` mines a NIP-13 proof of work into it first (blocks while mining, exit 124 on timeout with nothing published; `--json` adds `pow`, `pow_target`, `pow_millis`). |
|
||||
| `amy notes post TEXT [--relay URL] [--pow BITS [--pow-timeout SECS]]` | Publish a kind:1 short text note; `--pow` mines a NIP-13 proof of work into it first, using all cores (blocks while mining, exit 124 on timeout with nothing published; `--json` adds `pow`, `pow_target`, `pow_millis`). |
|
||||
| `amy notes feed [--author USER \| --following] [--limit N]` | Read recent kind:1 notes (yours, one user's, or your follow set). |
|
||||
| `amy profile show [USER]` | Print kind:0 metadata. USER accepts npub/nprofile/hex/NIP-05; defaults to self. |
|
||||
| `amy profile edit --name … --about … --picture URL …` | Patch and re-publish your kind:0. |
|
||||
|
||||
@@ -81,13 +81,14 @@ object PostCommand {
|
||||
var powMillis: Long? = null
|
||||
val readyToSign =
|
||||
if (powTarget != null) {
|
||||
System.err.println("mining $powTarget bits…")
|
||||
val threads = Runtime.getRuntime().availableProcessors().coerceAtLeast(1)
|
||||
System.err.println("mining $powTarget bits… ($threads threads)")
|
||||
val deadlineNanos = powTimeoutSec?.let { System.nanoTime() + it * 1_000_000_000L }
|
||||
val startedAt = System.nanoTime()
|
||||
val mined =
|
||||
try {
|
||||
withContext(Dispatchers.Default) {
|
||||
PoWMiner.run(template, ctx.signer.pubKey, powTarget) {
|
||||
PoWMiner.mine(template, ctx.signer.pubKey, powTarget, threads) {
|
||||
deadlineNanos == null || System.nanoTime() < deadlineNanos
|
||||
}
|
||||
}
|
||||
|
||||
@@ -108,13 +108,18 @@ object PowCommands {
|
||||
rest: Array<String>,
|
||||
): Int {
|
||||
val args = Args(rest)
|
||||
val usage = "pow mine --target N [--pubkey HEX] [--timeout SECS] <template-json | ->"
|
||||
val usage = "pow mine --target N [--pubkey HEX] [--timeout SECS] [--threads N] <template-json | ->"
|
||||
|
||||
val target = args.flags["target"]?.toIntOrNull() ?: return Output.error("bad_args", usage)
|
||||
if (target < 1 || target > MAX_DIFFICULTY) {
|
||||
return Output.error("bad_args", "--target must be between 1 and $MAX_DIFFICULTY")
|
||||
}
|
||||
|
||||
val threads = args.intFlag("threads", defaultThreads())
|
||||
if (threads < 1) {
|
||||
return Output.error("bad_args", "--threads must be >= 1")
|
||||
}
|
||||
|
||||
val json = readPayload(args.positional.toTypedArray()) ?: return Output.error("bad_args", usage)
|
||||
val template =
|
||||
try {
|
||||
@@ -142,13 +147,13 @@ object PowCommands {
|
||||
val timeoutSec = args.flags["timeout"]?.toLongOrNull()
|
||||
val deadlineNanos = timeoutSec?.let { System.nanoTime() + it * 1_000_000_000L }
|
||||
|
||||
System.err.println("mining $target bits for ${pubKey.take(8)}…")
|
||||
System.err.println("mining $target bits for ${pubKey.take(8)}… ($threads threads)")
|
||||
val startedAt = System.nanoTime()
|
||||
|
||||
val mined =
|
||||
try {
|
||||
withContext(Dispatchers.Default) {
|
||||
PoWMiner.run(template, pubKey, target) {
|
||||
PoWMiner.mine(template, pubKey, target, threads) {
|
||||
deadlineNanos == null || System.nanoTime() < deadlineNanos
|
||||
}
|
||||
}
|
||||
@@ -176,27 +181,38 @@ object PowCommands {
|
||||
"pow" to PoWRankEvaluator.calculatePowRankOf(id),
|
||||
"pow_target" to target,
|
||||
"pow_millis" to elapsedMs,
|
||||
"threads" to threads,
|
||||
"template_json" to mined.toJson(),
|
||||
),
|
||||
)
|
||||
return 0
|
||||
}
|
||||
|
||||
/** `amy pow bench` — hash rate + expected mining time per common target. */
|
||||
/**
|
||||
* `amy pow bench` — single-core and all-cores hash rate, plus expected
|
||||
* mining time per common target at the all-cores rate (what `pow mine`
|
||||
* uses by default).
|
||||
*/
|
||||
private suspend fun bench(): Int {
|
||||
val rate = PoWEstimator.hashesPerSecond()
|
||||
val threads = defaultThreads()
|
||||
val singleRate = PoWEstimator.hashesPerSecond()
|
||||
val parallelRate = PoWEstimator.hashesPerSecond(threads)
|
||||
Output.emit(
|
||||
mapOf(
|
||||
"hashes_per_second" to rate.roundToLong(),
|
||||
"hashes_per_second" to parallelRate.roundToLong(),
|
||||
"hashes_per_second_single_core" to singleRate.roundToLong(),
|
||||
"threads" to threads,
|
||||
"expected_seconds" to
|
||||
listOf(16, 20, 24, 28).associate { bits ->
|
||||
bits.toString() to PoWEstimator.estimateSeconds(bits, rate)
|
||||
bits.toString() to PoWEstimator.estimateSeconds(bits, parallelRate)
|
||||
},
|
||||
),
|
||||
)
|
||||
return 0
|
||||
}
|
||||
|
||||
private fun defaultThreads(): Int = Runtime.getRuntime().availableProcessors().coerceAtLeast(1)
|
||||
|
||||
private fun readPayload(rest: Array<String>): String? {
|
||||
val arg = rest.firstOrNull() ?: return null
|
||||
val payload = if (arg == "-") System.`in`.readBytes().decodeToString() else arg
|
||||
|
||||
+39
-11
@@ -20,9 +20,12 @@
|
||||
*/
|
||||
package com.vitorpamplona.amethyst.commons.service.pow
|
||||
|
||||
import com.vitorpamplona.quartz.utils.sha256.sha256
|
||||
import com.vitorpamplona.quartz.utils.sha256.sha256Into
|
||||
import kotlinx.coroutines.CoroutineDispatcher
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.async
|
||||
import kotlinx.coroutines.awaitAll
|
||||
import kotlinx.coroutines.coroutineScope
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
import kotlinx.coroutines.withContext
|
||||
@@ -48,18 +51,41 @@ object PoWEstimator {
|
||||
private const val BATCH = 2_000
|
||||
private val BENCH_DURATION = 250.milliseconds
|
||||
|
||||
private var cachedRate: Double? = null
|
||||
// cached per worker count; guarded by benchLock (single-flight so
|
||||
// concurrent first callers — e.g. the settings screen recomposing while
|
||||
// the composer chip opens — share one ~250 ms benchmark instead of each
|
||||
// burning the cores).
|
||||
private val cachedRates = mutableMapOf<Int, Double>()
|
||||
private val benchLock = Mutex()
|
||||
|
||||
suspend fun hashesPerSecond(dispatcher: CoroutineDispatcher = Dispatchers.Default): Double =
|
||||
cachedRate ?: withContext(dispatcher) {
|
||||
// single-flight: concurrent first callers (e.g. the settings screen
|
||||
// recomposing while the composer chip opens) share one ~250 ms
|
||||
// benchmark instead of each burning a core.
|
||||
benchLock.withLock {
|
||||
cachedRate ?: benchmark().also { cachedRate = it }
|
||||
suspend fun hashesPerSecond(dispatcher: CoroutineDispatcher = Dispatchers.Default): Double = hashesPerSecond(1, dispatcher)
|
||||
|
||||
/**
|
||||
* Aggregate hash rate with [workers] concurrent miners — what
|
||||
* [com.vitorpamplona.quartz.nip13Pow.miner.PoWMiner.mine] achieves when
|
||||
* racing that many workers. Each worker runs its own ~250 ms benchmark
|
||||
* loop concurrently and the rates are summed, so contention between
|
||||
* cores is priced in.
|
||||
*/
|
||||
suspend fun hashesPerSecond(
|
||||
workers: Int,
|
||||
dispatcher: CoroutineDispatcher = Dispatchers.Default,
|
||||
): Double {
|
||||
val count = workers.coerceAtLeast(1)
|
||||
benchLock.withLock {
|
||||
cachedRates[count]?.let { return it }
|
||||
return withContext(dispatcher) {
|
||||
val rate =
|
||||
if (count == 1) {
|
||||
benchmark()
|
||||
} else {
|
||||
coroutineScope { List(count) { async { benchmark() } }.awaitAll().sum() }
|
||||
}
|
||||
cachedRates[count] = rate
|
||||
rate
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fun estimateSeconds(
|
||||
difficulty: Int,
|
||||
@@ -68,14 +94,16 @@ object PoWEstimator {
|
||||
|
||||
private fun benchmark(): Double {
|
||||
val payload = ByteArray(PAYLOAD_BYTES) { (it % 251).toByte() }
|
||||
// same allocation-free hashing the miner's hot loop uses
|
||||
val out = ByteArray(32)
|
||||
|
||||
// warm up JIT/caches so the measured window reflects steady state
|
||||
repeat(3 * BATCH) { sha256(payload) }
|
||||
repeat(3 * BATCH) { sha256Into(out, payload, payload.size) }
|
||||
|
||||
val mark = TimeSource.Monotonic.markNow()
|
||||
var count = 0L
|
||||
while (mark.elapsedNow() < BENCH_DURATION) {
|
||||
repeat(BATCH) { sha256(payload) }
|
||||
repeat(BATCH) { sha256Into(out, payload, payload.size) }
|
||||
count += BATCH
|
||||
}
|
||||
return count / mark.elapsedNow().toDouble(DurationUnit.SECONDS)
|
||||
|
||||
+7
@@ -86,6 +86,13 @@ object PoWPolicy {
|
||||
*/
|
||||
const val MAX_DIFFICULTY = 40
|
||||
|
||||
/**
|
||||
* How many parallel miner workers one mining job should use on a device
|
||||
* with [availableProcessors] cores: half of them, so a running miner never
|
||||
* starves the UI/render threads or the relay client. At least 1.
|
||||
*/
|
||||
fun minerWorkers(availableProcessors: Int): Int = (availableProcessors / 2).coerceAtLeast(1)
|
||||
|
||||
private const val LONG_FORM_DRAFT_KIND = 30024
|
||||
|
||||
/** NIP-51 sets and other settings-like addressable kinds. */
|
||||
|
||||
+6
-2
@@ -96,7 +96,10 @@ data class PoWJobFailure(
|
||||
* to the normal sign+broadcast continuation captured at enqueue time.
|
||||
*
|
||||
* FIFO with at most [maxConcurrent] concurrent miners so a burst of posts
|
||||
* queues up instead of spawning unbounded CPU work. Jobs are cancellable
|
||||
* queues up instead of spawning unbounded CPU work. Each job's nonce search
|
||||
* runs on [minerThreads] parallel workers (see PoWMiner.mine) — callers that
|
||||
* bring their own mine lambda via [enqueueStaged]/[enqueueWork] should reuse
|
||||
* this value to stay inside the queue's CPU budget. Jobs are cancellable
|
||||
* while QUEUED or MINING; once PUBLISHING starts, cancel is a no-op.
|
||||
*
|
||||
* Template jobs enqueued with a [PersistedPoWJob] record are checkpointed to
|
||||
@@ -115,6 +118,7 @@ data class PoWJobFailure(
|
||||
class PoWPublishQueue(
|
||||
private val scope: CoroutineScope,
|
||||
maxConcurrent: Int = 1,
|
||||
val minerThreads: Int = 1,
|
||||
miningDispatcher: CoroutineDispatcher = Dispatchers.Default,
|
||||
private val persistence: PoWJobPersistence? = null,
|
||||
private val onQueueActive: () -> Unit = {},
|
||||
@@ -198,7 +202,7 @@ class PoWPublishQueue(
|
||||
} else {
|
||||
template
|
||||
}
|
||||
PoWMiner.run(toMine, pubKey, difficulty, isActive)
|
||||
PoWMiner.mine(toMine, pubKey, difficulty, minerThreads, isActive)
|
||||
},
|
||||
publish = onMined,
|
||||
)
|
||||
|
||||
+10
@@ -28,6 +28,16 @@ import kotlin.test.assertTrue
|
||||
class PoWPolicyTest {
|
||||
val allCategories = PoWCategory.entries.toSet()
|
||||
|
||||
@Test
|
||||
fun minerWorkersIsHalfTheCoresButNeverZero() {
|
||||
assertEquals(1, PoWPolicy.minerWorkers(1))
|
||||
assertEquals(1, PoWPolicy.minerWorkers(2))
|
||||
assertEquals(2, PoWPolicy.minerWorkers(4))
|
||||
assertEquals(3, PoWPolicy.minerWorkers(6))
|
||||
assertEquals(4, PoWPolicy.minerWorkers(8))
|
||||
assertEquals(1, PoWPolicy.minerWorkers(0))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun difficultyOffMinesNothing() {
|
||||
assertNull(PoWPolicy.shouldMine(1, 0, allCategories))
|
||||
|
||||
@@ -25,19 +25,30 @@ import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
||||
import com.vitorpamplona.quartz.nip01Core.crypto.EventHasherSerializer
|
||||
import com.vitorpamplona.quartz.nip01Core.signers.EventTemplate
|
||||
import com.vitorpamplona.quartz.nip13Pow.tags.PoWTag
|
||||
import com.vitorpamplona.quartz.utils.sha256.sha256
|
||||
import com.vitorpamplona.quartz.utils.sha256.sha256Into
|
||||
import kotlinx.coroutines.CompletableDeferred
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.coroutineScope
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlin.coroutines.cancellation.CancellationException
|
||||
|
||||
class PoWMiner(
|
||||
val buffer: MiningBuffer,
|
||||
val desiredPoW: Int,
|
||||
val isActive: () -> Boolean = { true },
|
||||
// first nonce byte the search is allowed to change; bytes between
|
||||
// nonceStarts and this index stay fixed (a parallel worker's prefix).
|
||||
val searchFrom: Int = buffer.nonceStarts,
|
||||
) {
|
||||
val emptyBytesForDesiredPoW = desiredPoW / 8
|
||||
|
||||
fun reachedDesiredPoW(byteArray: ByteArray) = PoWRankEvaluator.atLeastPowRank(sha256(byteArray), desiredPoW, emptyBytesForDesiredPoW)
|
||||
// sha256Into writes every attempt's hash here instead of allocating a fresh
|
||||
// 32-byte array per hash, keeping the hot loop allocation-free.
|
||||
private val hashOut = ByteArray(32)
|
||||
|
||||
fun run() = runDigit(buffer.nonceStarts)
|
||||
fun reachedDesiredPoW(byteArray: ByteArray) = PoWRankEvaluator.atLeastPowRank(sha256Into(hashOut, byteArray, byteArray.size), desiredPoW, emptyBytesForDesiredPoW)
|
||||
|
||||
fun run() = runDigit(searchFrom)
|
||||
|
||||
private fun runDigit(index: Int): Boolean {
|
||||
// checks once every VALID_BYTES.size^2 hashes: cheap enough to not slow
|
||||
@@ -74,6 +85,10 @@ class PoWMiner(
|
||||
* The miner creates a stringified json template and changes the nonce directly in the UTF-8 ByteArray representation
|
||||
* to avoid having to recompute the json objects and stringify it.
|
||||
*
|
||||
* The search enumerates the nonce space deterministically ([VALID_BYTES]
|
||||
* in order at every position), so for a given template and pubkey every
|
||||
* call hashes the same sequence of candidates.
|
||||
*
|
||||
* [isActive] is polled while mining; returning false aborts the search with a
|
||||
* [CancellationException] so callers can cancel long-running jobs cooperatively.
|
||||
*/
|
||||
@@ -82,6 +97,19 @@ class PoWMiner(
|
||||
pubKey: HexKey,
|
||||
desiredPoW: Int,
|
||||
isActive: () -> Boolean = { true },
|
||||
): EventTemplate<T> = search(template, pubKey, desiredPoW, isActive, "")
|
||||
|
||||
/**
|
||||
* [noncePrefix] is kept verbatim at the front of the nonce while only the
|
||||
* bytes after it are enumerated — parallel workers get distinct prefixes
|
||||
* so their search spaces never overlap.
|
||||
*/
|
||||
private fun <T : Event> search(
|
||||
template: EventTemplate<T>,
|
||||
pubKey: HexKey,
|
||||
desiredPoW: Int,
|
||||
isActive: () -> Boolean,
|
||||
noncePrefix: String,
|
||||
): EventTemplate<T> {
|
||||
// sha256 ids have 256 bits; anything outside would index past the
|
||||
// hash (or never terminate) deep inside the hot loop.
|
||||
@@ -90,7 +118,7 @@ class PoWMiner(
|
||||
var nextSize = STARTING_NONCE_SIZE
|
||||
|
||||
do {
|
||||
val initialNonce = randomBase(nextSize)
|
||||
val initialNonce = noncePrefix + randomBase(nextSize)
|
||||
|
||||
val bytes =
|
||||
EventHasherSerializer
|
||||
@@ -104,9 +132,9 @@ class PoWMiner(
|
||||
|
||||
val startIndex = bytes.indexOf(initialNonce.encodeToByteArray())
|
||||
|
||||
val buffer = MiningBuffer(bytes, startIndex, startIndex + nextSize)
|
||||
val buffer = MiningBuffer(bytes, startIndex, startIndex + initialNonce.length)
|
||||
|
||||
if (PoWMiner(buffer, desiredPoW, isActive).run()) {
|
||||
if (PoWMiner(buffer, desiredPoW, isActive, startIndex + noncePrefix.length).run()) {
|
||||
return EventTemplate(
|
||||
template.createdAt,
|
||||
template.kind,
|
||||
@@ -120,5 +148,79 @@ class PoWMiner(
|
||||
|
||||
throw RuntimeException("Could not find PoW")
|
||||
}
|
||||
|
||||
/**
|
||||
* Distinct fixed nonce prefix for each racing worker, encoding the worker
|
||||
* index in base-[VALID_CHARS].size at the smallest width that fits
|
||||
* [workers] — one char up to 73 workers.
|
||||
*/
|
||||
private fun workerPrefix(
|
||||
worker: Int,
|
||||
workers: Int,
|
||||
): String {
|
||||
var width = 1
|
||||
var capacity = VALID_CHARS.size
|
||||
while (capacity < workers) {
|
||||
width++
|
||||
capacity *= VALID_CHARS.size
|
||||
}
|
||||
|
||||
var remaining = worker
|
||||
val prefix = CharArray(width)
|
||||
for (i in width - 1 downTo 0) {
|
||||
prefix[i] = VALID_CHARS[remaining % VALID_CHARS.size]
|
||||
remaining /= VALID_CHARS.size
|
||||
}
|
||||
return prefix.concatToString()
|
||||
}
|
||||
|
||||
/**
|
||||
* Multi-core variant of [run]: races [workers] searches over disjoint
|
||||
* slices of the nonce space (each worker's nonce carries a distinct fixed
|
||||
* prefix) and returns the first template to reach [desiredPoW]. Workers
|
||||
* share nothing but the finish flag, so the hash rate scales roughly
|
||||
* linearly with cores.
|
||||
*
|
||||
* Throws the same [CancellationException] as [run] when [isActive] flips
|
||||
* false before a nonce is found.
|
||||
*/
|
||||
suspend fun <T : Event> mine(
|
||||
template: EventTemplate<T>,
|
||||
pubKey: HexKey,
|
||||
desiredPoW: Int,
|
||||
workers: Int = 1,
|
||||
isActive: () -> Boolean = { true },
|
||||
): EventTemplate<T> {
|
||||
require(workers >= 1) { "workers must be >= 1, was $workers" }
|
||||
if (workers == 1) return run(template, pubKey, desiredPoW, isActive)
|
||||
|
||||
return coroutineScope {
|
||||
val winner = CompletableDeferred<EventTemplate<T>>()
|
||||
val race =
|
||||
launch {
|
||||
repeat(workers) { worker ->
|
||||
launch(Dispatchers.Default) {
|
||||
winner.complete(
|
||||
search(template, pubKey, desiredPoW, {
|
||||
isActive() && !winner.isCompleted
|
||||
}, workerPrefix(worker, workers)),
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
// every worker aborting without a win means the caller's isActive
|
||||
// flipped: rethrow the CancellationException the workers swallowed
|
||||
// (a losing worker's complete() is a no-op, so this can't clobber
|
||||
// a real result).
|
||||
race.invokeOnCompletion {
|
||||
winner.completeExceptionally(CancellationException("PoW mining was cancelled"))
|
||||
}
|
||||
try {
|
||||
winner.await()
|
||||
} finally {
|
||||
race.cancel()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+6
-1
@@ -38,12 +38,16 @@ import com.vitorpamplona.quartz.nip57Zaps.LnZapRequestEvent
|
||||
* Only events whose kind is in [kindsToMine] are mined; anything else (e.g. the
|
||||
* seal and rumor of a gift-wrapped flow) passes through untouched. Templates that
|
||||
* already carry a nonce tag are not mined again.
|
||||
*
|
||||
* [workers] parallel searches race over disjoint nonce slices (see
|
||||
* [PoWMiner.mine]); 1 keeps the historical single-threaded behavior.
|
||||
*/
|
||||
class PoWNostrSigner(
|
||||
val signer: NostrSigner,
|
||||
val desiredPoW: Int,
|
||||
val kindsToMine: Set<Int>,
|
||||
val isActive: () -> Boolean = { true },
|
||||
val workers: Int = 1,
|
||||
) : NostrSigner(signer.pubKey) {
|
||||
override fun isWriteable(): Boolean = signer.isWriteable()
|
||||
|
||||
@@ -55,10 +59,11 @@ class PoWNostrSigner(
|
||||
): T =
|
||||
if (kind in kindsToMine && tags.none { PoWTag.hasTagWithContent(it) }) {
|
||||
val mined =
|
||||
PoWMiner.run(
|
||||
PoWMiner.mine(
|
||||
template = EventTemplate<T>(createdAt, kind, tags, content),
|
||||
pubKey = pubKey,
|
||||
desiredPoW = desiredPoW,
|
||||
workers = workers,
|
||||
isActive = isActive,
|
||||
)
|
||||
signer.sign(mined.createdAt, mined.kind, mined.tags, mined.content)
|
||||
|
||||
+91
@@ -0,0 +1,91 @@
|
||||
/*
|
||||
* Copyright (c) 2025 Vitor Pamplona
|
||||
*
|
||||
* Permission is hereby granted, free of charge, to any person obtaining a copy of
|
||||
* this software and associated documentation files (the "Software"), to deal in
|
||||
* the Software without restriction, including without limitation the rights to use,
|
||||
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
|
||||
* Software, and to permit persons to whom the Software is furnished to do so,
|
||||
* subject to the following conditions:
|
||||
*
|
||||
* The above copyright notice and this permission notice shall be included in all
|
||||
* copies or substantial portions of the Software.
|
||||
*
|
||||
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
|
||||
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
|
||||
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
|
||||
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
|
||||
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
*/
|
||||
package com.vitorpamplona.quartz.nip13Pow
|
||||
|
||||
import com.vitorpamplona.quartz.nip01Core.crypto.EventHasherSerializer
|
||||
import com.vitorpamplona.quartz.nip01Core.signers.EventTemplate
|
||||
import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
|
||||
import com.vitorpamplona.quartz.nip13Pow.miner.PoWMiner
|
||||
import com.vitorpamplona.quartz.nip13Pow.miner.PoWRankEvaluator
|
||||
import com.vitorpamplona.quartz.nip13Pow.tags.PoWTag
|
||||
import com.vitorpamplona.quartz.utils.sha256.sha256
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import kotlin.coroutines.cancellation.CancellationException
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertFailsWith
|
||||
import kotlin.test.assertNotNull
|
||||
import kotlin.test.assertTrue
|
||||
import kotlin.time.Duration.Companion.milliseconds
|
||||
import kotlin.time.TimeSource
|
||||
|
||||
class PoWMinerParallelTest {
|
||||
val pubKey = "460c25e682fda7832b52d1f22d3d22b3176d972f60dcdc3212ed8c92ef85065c"
|
||||
|
||||
val baseTemplate =
|
||||
EventTemplate<TextNoteEvent>(
|
||||
1683596206,
|
||||
TextNoteEvent.KIND,
|
||||
emptyArray(),
|
||||
"A note to mine",
|
||||
)
|
||||
|
||||
@Test
|
||||
fun parallelMinerFindsAValidPoW() =
|
||||
runTest {
|
||||
val desiredPoW = 12
|
||||
val mined = PoWMiner.mine(baseTemplate, pubKey, desiredPoW, workers = 4)
|
||||
|
||||
val powTag = mined.tags.firstNotNullOfOrNull { PoWTag.parse(it) }
|
||||
assertNotNull(powTag, "mined template must carry a nonce tag")
|
||||
|
||||
val id =
|
||||
sha256(
|
||||
EventHasherSerializer.fastMakeJsonForId(
|
||||
pubKey = pubKey,
|
||||
createdAt = mined.createdAt,
|
||||
kind = mined.kind,
|
||||
tags = mined.tags,
|
||||
content = mined.content,
|
||||
),
|
||||
)
|
||||
|
||||
assertTrue(
|
||||
PoWRankEvaluator.atLeastPowRank(id, desiredPoW, desiredPoW / 8),
|
||||
"mined id must reach the desired PoW",
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun cancellationAbortsAllParallelWorkers() =
|
||||
runTest {
|
||||
val start = TimeSource.Monotonic.markNow()
|
||||
// 256 bits never completes; only the isActive deadline can end the run.
|
||||
assertFailsWith<CancellationException> {
|
||||
PoWMiner.mine(baseTemplate, pubKey, 256, workers = 4) {
|
||||
start.elapsedNow() < 150.milliseconds
|
||||
}
|
||||
}
|
||||
assertTrue(
|
||||
start.elapsedNow() < 5_000.milliseconds,
|
||||
"all workers must stop shortly after isActive flips, took ${start.elapsedNow()}",
|
||||
)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user