diff --git a/components/stratum_v2/include/sv2_protocol.h b/components/stratum_v2/include/sv2_protocol.h index 6f34744021..dae465f6c3 100644 --- a/components/stratum_v2/include/sv2_protocol.h +++ b/components/stratum_v2/include/sv2_protocol.h @@ -84,6 +84,9 @@ typedef struct { uint16_t coinbase_prefix_len; uint8_t *coinbase_suffix; // heap uint16_t coinbase_suffix_len; + uint8_t extranonce_prefix[32]; + uint8_t extranonce_prefix_len; + uint8_t extranonce_size; } sv2_ext_job_t; #define SV2_PENDING_JOBS_SIZE 8 diff --git a/main/self_test/self_test.c b/main/self_test/self_test.c index 0335d9a48a..14e1b9bf2d 100644 --- a/main/self_test/self_test.c +++ b/main/self_test/self_test.c @@ -49,6 +49,11 @@ static const char * TAG = "self_test"; static SemaphoreHandle_t longPressSemaphore; static bool isFactoryTest = false; +static void self_test_free_v1_work(void *work) +{ + STRATUM_V1_free_mining_notify((mining_notify *)work); +} + // local function prototypes static void tests_done(GlobalState * GLOBAL_STATE, bool test_result); @@ -527,7 +532,12 @@ void self_test_task(void * pvParameters) if (msg.method == MINING_NOTIFY) { ESP_LOGI(TAG, "Enqueuing mock work into stratum_queue"); - queue_enqueue(&GLOBAL_STATE->stratum_queue, msg.mining_notification); + queue_set_source(&GLOBAL_STATE->stratum_queue, WORK_ITEM_STRATUM_V1); + queue_enqueue( + &GLOBAL_STATE->stratum_queue, + work_queue_item_create(&GLOBAL_STATE->stratum_queue, + msg.mining_notification, WORK_ITEM_STRATUM_V1, + self_test_free_v1_work)); } else { ESP_LOGE(TAG, "Failed to parse mock mining notification"); tests_done(GLOBAL_STATE, false); diff --git a/main/system.c b/main/system.c index d463902535..6d3e15c422 100644 --- a/main/system.c +++ b/main/system.c @@ -331,10 +331,9 @@ esp_err_t SYSTEM_init_peripherals(GlobalState * GLOBAL_STATE) { return ESP_OK; } -void SYSTEM_clean_jobs_queue(GlobalState * GLOBAL_STATE) +static void system_reset_work_state(GlobalState *GLOBAL_STATE, work_item_kind_t kind) { - ESP_LOGI(TAG, "Clean Jobs: clearing queue"); - queue_clear(&GLOBAL_STATE->stratum_queue); + queue_set_source(&GLOBAL_STATE->stratum_queue, kind); pthread_mutex_lock(&GLOBAL_STATE->valid_jobs_lock); for (int i = 0; i < 128; i = i + 4) { @@ -346,6 +345,18 @@ void SYSTEM_clean_jobs_queue(GlobalState * GLOBAL_STATE) hashrate_monitor_reset_measurements(GLOBAL_STATE); } +void SYSTEM_clean_jobs_queue(GlobalState *GLOBAL_STATE) +{ + ESP_LOGI(TAG, "Clean Jobs: clearing queue"); + system_reset_work_state(GLOBAL_STATE, WORK_ITEM_NONE); +} + +void SYSTEM_set_work_source(GlobalState *GLOBAL_STATE, work_item_kind_t kind) +{ + ESP_LOGI(TAG, "Activating work source %d", kind); + system_reset_work_state(GLOBAL_STATE, kind); +} + void SYSTEM_notify_accepted_share(GlobalState * GLOBAL_STATE) { SystemModule * module = &GLOBAL_STATE->SYSTEM_MODULE; diff --git a/main/system.h b/main/system.h index fda2d0f499..a69b5d184c 100644 --- a/main/system.h +++ b/main/system.h @@ -4,6 +4,7 @@ #include "esp_err.h" #include "sv2_protocol.h" +#include "work_queue.h" typedef struct GlobalState GlobalState; @@ -25,6 +26,7 @@ esp_err_t SYSTEM_init_peripherals(GlobalState * GLOBAL_STATE); // and reset hashrate measurements so reconnects don't spike the average. // Shared by the SV1 and SV2 tasks. void SYSTEM_clean_jobs_queue(GlobalState * GLOBAL_STATE); +void SYSTEM_set_work_source(GlobalState *GLOBAL_STATE, work_item_kind_t kind); void SYSTEM_notify_accepted_share(GlobalState * GLOBAL_STATE); void SYSTEM_notify_rejected_share(GlobalState * GLOBAL_STATE, char * error_msg); diff --git a/main/tasks/create_jobs_task.c b/main/tasks/create_jobs_task.c index 57e9668cbd..e1259941b8 100644 --- a/main/tasks/create_jobs_task.c +++ b/main/tasks/create_jobs_task.c @@ -26,21 +26,27 @@ static void generate_work(GlobalState *GLOBAL_STATE, mining_notify *notification static void generate_work_sv2(GlobalState *GLOBAL_STATE, sv2_job_t *job, double difficulty); static void generate_work_sv2_ext(GlobalState *GLOBAL_STATE, sv2_ext_job_t *job, double difficulty, uint64_t extranonce_2_counter); -// Free a work item using the correct free function for the protocol it was created under -static void free_work_item(GlobalState *GLOBAL_STATE, void *work, stratum_protocol_t protocol) +static const char *work_item_kind_name(work_item_kind_t kind) { - if (!work) return; - if (protocol == STRATUM_PROTOCOL_V2) { - if (stratum_v2_is_extended_channel(GLOBAL_STATE)) { - sv2_ext_job_free((sv2_ext_job_t *)work); - } else { - free(work); // sv2_job_t is flat - } - } else { - STRATUM_V1_free_mining_notify(work); + switch (kind) { + case WORK_ITEM_STRATUM_V1: + return "Stratum V1"; + case WORK_ITEM_STRATUM_V2_STANDARD: + return "SV2 standard"; + case WORK_ITEM_STRATUM_V2_EXTENDED: + return "SV2 extended"; + default: + return "none"; } } +static bool work_item_matches_source(const work_queue_item_t *item, + work_queue_source_t source) +{ + return item->data != NULL && item->kind == source.kind && + item->source_epoch == source.epoch; +} + void create_jobs_task(void *pvParameters) { GlobalState *GLOBAL_STATE = (GlobalState *)pvParameters; @@ -54,8 +60,7 @@ void create_jobs_task(void *pvParameters) } double difficulty = GLOBAL_STATE->pool_difficulty; - void *current_work = NULL; - stratum_protocol_t current_work_protocol = GLOBAL_STATE->stratum_protocol; + work_queue_item_t current_work = {0}; uint64_t extranonce_2 = 0; int timeout_ms = ASIC_get_asic_job_frequency_ms(GLOBAL_STATE); @@ -63,56 +68,52 @@ void create_jobs_task(void *pvParameters) ESP_LOGI(TAG, "ASIC Ready!"); while (1) { - // Read protocol dynamically each iteration (coordinator may have switched it) - stratum_protocol_t active_protocol = GLOBAL_STATE->stratum_protocol; - - // If protocol changed, discard current_work (it belongs to the old protocol) - // Always update current_work_protocol so the post-dequeue check doesn't - // incorrectly discard the first valid work item from the new protocol. - if (active_protocol != current_work_protocol) { - if (current_work != NULL) { - ESP_LOGI(TAG, "Protocol switched from %s to %s, discarding current work", - current_work_protocol == STRATUM_PROTOCOL_V2 ? STRATUM_V2 : STRATUM_V1, - active_protocol == STRATUM_PROTOCOL_V2 ? STRATUM_V2 : STRATUM_V1); - free_work_item(GLOBAL_STATE, current_work, current_work_protocol); - current_work = NULL; - } - current_work_protocol = active_protocol; + work_queue_source_t source = queue_get_source(&GLOBAL_STATE->stratum_queue); + + if (current_work.data != NULL && + !work_item_matches_source(¤t_work, source)) { + ESP_LOGI(TAG, "Work source switched from %s to %s, discarding current work", + work_item_kind_name(current_work.kind), + work_item_kind_name(source.kind)); + work_queue_item_free(¤t_work); } uint64_t start_time = esp_timer_get_time(); - void *new_work = queue_dequeue_timeout(&GLOBAL_STATE->stratum_queue, timeout_ms); + work_queue_item_t new_work = + queue_dequeue_timeout(&GLOBAL_STATE->stratum_queue, timeout_ms); timeout_ms -= (esp_timer_get_time() - start_time) / 1000; - if (new_work != NULL) { - active_protocol = GLOBAL_STATE->stratum_protocol; - - // Free previous work using the protocol it was created under - free_work_item(GLOBAL_STATE, current_work, current_work_protocol); - current_work = NULL; - - if (active_protocol != current_work_protocol) { - // Protocol switched during our blocking dequeue. - // The dequeued item may be from either the old or new protocol — - // we cannot safely determine which type it is, so discard it. - // free() is safe for both sv2_job_t (flat) and mining_notify (malloc'd; - // internal strings leak but this is a rare protocol-switch event). - ESP_LOGW(TAG, "Protocol switch detected during dequeue, discarding stale item"); - free(new_work); - current_work_protocol = active_protocol; + if (new_work.data != NULL) { + source = queue_get_source(&GLOBAL_STATE->stratum_queue); + + if (!work_item_matches_source(&new_work, source)) { + ESP_LOGW(TAG, "Discarding stale %s work while %s is active", + work_item_kind_name(new_work.kind), + work_item_kind_name(source.kind)); + work_queue_item_free(&new_work); timeout_ms = ASIC_get_asic_job_frequency_ms(GLOBAL_STATE); continue; } - // Protocol unchanged — item matches current_work_protocol. Safe to cast. - if (current_work_protocol == STRATUM_PROTOCOL_V2) { - if (stratum_v2_is_extended_channel(GLOBAL_STATE)) { - ESP_LOGI(TAG, "New Work Dequeued SV2 ext job %lu", ((sv2_ext_job_t *)new_work)->job_id); - } else { - ESP_LOGI(TAG, "New Work Dequeued SV2 job %lu", ((sv2_job_t *)new_work)->job_id); - } - } else { - ESP_LOGI(TAG, "New Work Dequeued %s", ((mining_notify *)new_work)->job_id); + work_queue_item_free(¤t_work); + + switch (new_work.kind) { + case WORK_ITEM_STRATUM_V2_EXTENDED: + ESP_LOGI(TAG, "New Work Dequeued SV2 ext job %lu", + ((sv2_ext_job_t *)new_work.data)->job_id); + break; + case WORK_ITEM_STRATUM_V2_STANDARD: + ESP_LOGI(TAG, "New Work Dequeued SV2 job %lu", + ((sv2_job_t *)new_work.data)->job_id); + break; + case WORK_ITEM_STRATUM_V1: + ESP_LOGI(TAG, "New Work Dequeued %s", + ((mining_notify *)new_work.data)->job_id); + break; + default: + work_queue_item_free(&new_work); + timeout_ms = ASIC_get_asic_job_frequency_ms(GLOBAL_STATE); + continue; } current_work = new_work; @@ -133,20 +134,25 @@ void create_jobs_task(void *pvParameters) // Check clean_jobs flag bool clean; - if (current_work_protocol == STRATUM_PROTOCOL_V2) { - if (stratum_v2_is_extended_channel(GLOBAL_STATE)) { - clean = ((sv2_ext_job_t *)current_work)->clean_jobs; - } else { - clean = ((sv2_job_t *)current_work)->clean_jobs; - } - } else { - clean = ((mining_notify *)current_work)->clean_jobs; + switch (current_work.kind) { + case WORK_ITEM_STRATUM_V2_EXTENDED: + clean = ((sv2_ext_job_t *)current_work.data)->clean_jobs; + break; + case WORK_ITEM_STRATUM_V2_STANDARD: + clean = ((sv2_job_t *)current_work.data)->clean_jobs; + break; + case WORK_ITEM_STRATUM_V1: + clean = ((mining_notify *)current_work.data)->clean_jobs; + break; + default: + work_queue_item_free(¤t_work); + continue; } if (!clean) { continue; } } else { - if (current_work == NULL) { + if (current_work.data == NULL) { vTaskDelay(100 / portTICK_PERIOD_MS); continue; } @@ -155,34 +161,38 @@ void create_jobs_task(void *pvParameters) // Re-sending the same job restarts the nonce search from 0 and // produces duplicate shares. Only send work on new jobs. // (V1 and SV2 extended are fine — extranonce_2 gives unique work each time.) - if (active_protocol == STRATUM_PROTOCOL_V2 && !stratum_v2_is_extended_channel(GLOBAL_STATE)) { + if (current_work.kind == WORK_ITEM_STRATUM_V2_STANDARD) { timeout_ms = ASIC_get_asic_job_frequency_ms(GLOBAL_STATE); continue; } } - // Final protocol check before generating work — protocol may have switched - // during a timeout dequeue while we still hold stale current_work - active_protocol = GLOBAL_STATE->stratum_protocol; - if (active_protocol != current_work_protocol) { - free_work_item(GLOBAL_STATE, current_work, current_work_protocol); - current_work = NULL; - current_work_protocol = active_protocol; + // The source may have switched during the dequeue or job preparation. + source = queue_get_source(&GLOBAL_STATE->stratum_queue); + if (!work_item_matches_source(¤t_work, source)) { + work_queue_item_free(¤t_work); timeout_ms = ASIC_get_asic_job_frequency_ms(GLOBAL_STATE); continue; } - // Generate and send job - if (active_protocol == STRATUM_PROTOCOL_V2) { - if (stratum_v2_is_extended_channel(GLOBAL_STATE)) { - generate_work_sv2_ext(GLOBAL_STATE, (sv2_ext_job_t *)current_work, difficulty, extranonce_2); + // Generate according to the immutable allocation type, never mutable + // protocol/channel state. + switch (current_work.kind) { + case WORK_ITEM_STRATUM_V2_EXTENDED: + generate_work_sv2_ext(GLOBAL_STATE, (sv2_ext_job_t *)current_work.data, + difficulty, extranonce_2); extranonce_2++; - } else { - generate_work_sv2(GLOBAL_STATE, (sv2_job_t *)current_work, difficulty); - } - } else { - generate_work(GLOBAL_STATE, (mining_notify *)current_work, extranonce_2, difficulty); - extranonce_2++; + break; + case WORK_ITEM_STRATUM_V2_STANDARD: + generate_work_sv2(GLOBAL_STATE, (sv2_job_t *)current_work.data, difficulty); + break; + case WORK_ITEM_STRATUM_V1: + generate_work(GLOBAL_STATE, (mining_notify *)current_work.data, + extranonce_2, difficulty); + extranonce_2++; + break; + default: + break; } timeout_ms = ASIC_get_asic_job_frequency_ms(GLOBAL_STATE); } @@ -309,8 +319,11 @@ static void generate_work_sv2(GlobalState *GLOBAL_STATE, sv2_job_t *sv2_job, dou static void generate_work_sv2_ext(GlobalState *GLOBAL_STATE, sv2_ext_job_t *ext_job, double difficulty, uint64_t extranonce_2_counter) { - sv2_conn_t *conn = GLOBAL_STATE->sv2_conn; - if (!conn) return; + if (ext_job->extranonce_size < 2 || ext_job->extranonce_size > 32 || + ext_job->extranonce_prefix_len > sizeof(ext_job->extranonce_prefix)) { + ESP_LOGE(TAG, "Invalid extended-channel extranonce parameters"); + return; + } bm_job *next_job = malloc(sizeof(bm_job)); if (!next_job) { @@ -322,7 +335,7 @@ static void generate_work_sv2_ext(GlobalState *GLOBAL_STATE, sv2_ext_job_t *ext_ // Derive extranonce_2 from counter // SV2 spec: extranonce_size is the miner's rollable portion (not total) - uint8_t extranonce_2_len = conn->extranonce_size; + uint8_t extranonce_2_len = ext_job->extranonce_size; uint8_t extranonce_2[32]; memset(extranonce_2, 0, sizeof(extranonce_2)); // Encode counter as big-endian bytes @@ -335,7 +348,7 @@ static void generate_work_sv2_ext(GlobalState *GLOBAL_STATE, sv2_ext_job_t *ext_ uint8_t coinbase_tx_hash[32]; calculate_coinbase_tx_hash_bin( ext_job->coinbase_prefix, ext_job->coinbase_prefix_len, - conn->extranonce_prefix, conn->extranonce_prefix_len, + ext_job->extranonce_prefix, ext_job->extranonce_prefix_len, extranonce_2, extranonce_2_len, ext_job->coinbase_suffix, ext_job->coinbase_suffix_len, coinbase_tx_hash); diff --git a/main/tasks/stratum_v1_task.c b/main/tasks/stratum_v1_task.c index 15799b488b..7a9d24035e 100644 --- a/main/tasks/stratum_v1_task.c +++ b/main/tasks/stratum_v1_task.c @@ -46,6 +46,11 @@ static const char *TAG = "stratum_v1_task"; static StratumApiV1Message stratum_api_v1_message = {}; +static void stratum_v1_free_work(void *work) +{ + STRATUM_V1_free_mining_notify((mining_notify *)work); +} + static int stratum_get_next_uid(GlobalState * GLOBAL_STATE) { taskENTER_CRITICAL(&GLOBAL_STATE->stratum_mux); @@ -176,9 +181,6 @@ void stratum_v1_task(void *pvParameters) char *stratum_url = GLOBAL_STATE->SYSTEM_MODULE.pools[pool_idx].url; uint16_t port = GLOBAL_STATE->SYSTEM_MODULE.pools[pool_idx].port; - // Set V1-specific free function for the work queue - GLOBAL_STATE->stratum_queue.free_fn = (void (*)(void *))STRATUM_V1_free_mining_notify; - STRATUM_V1_initialize_buffer(); int retry_attempts = 0; int retry_critical_attempts = 0; @@ -283,7 +285,7 @@ void stratum_v1_task(void *pvParameters) "%s%s", protocol, tls_status); stratum_v1_reset_uid(GLOBAL_STATE); - SYSTEM_clean_jobs_queue(GLOBAL_STATE); + SYSTEM_set_work_source(GLOBAL_STATE, WORK_ITEM_STRATUM_V1); ///// Start Stratum Action // mining.configure - ID: 1 @@ -344,15 +346,14 @@ void stratum_v1_task(void *pvParameters) case MINING_NOTIFY: GLOBAL_STATE->SYSTEM_MODULE.work_received++; SYSTEM_notify_new_ntime(GLOBAL_STATE, stratum_api_v1_message.mining_notification->ntime); - if (stratum_api_v1_message.mining_notification->clean_jobs && - (GLOBAL_STATE->stratum_queue.count > 0)) { - SYSTEM_clean_jobs_queue(GLOBAL_STATE); - } - if (GLOBAL_STATE->stratum_queue.count == QUEUE_SIZE) { - mining_notify *next_notify_json_str = (mining_notify *) queue_dequeue(&GLOBAL_STATE->stratum_queue); - STRATUM_V1_free_mining_notify(next_notify_json_str); + if (stratum_api_v1_message.mining_notification->clean_jobs) { + SYSTEM_set_work_source(GLOBAL_STATE, WORK_ITEM_STRATUM_V1); } - queue_enqueue(&GLOBAL_STATE->stratum_queue, stratum_api_v1_message.mining_notification); + queue_enqueue( + &GLOBAL_STATE->stratum_queue, + work_queue_item_create(&GLOBAL_STATE->stratum_queue, + stratum_api_v1_message.mining_notification, + WORK_ITEM_STRATUM_V1, stratum_v1_free_work)); decode_mining_notification(GLOBAL_STATE, stratum_api_v1_message.mining_notification); stratum_api_v1_message.mining_notification = NULL; break; diff --git a/main/tasks/stratum_v2_task.c b/main/tasks/stratum_v2_task.c index 95030a11e3..644dad11de 100644 --- a/main/tasks/stratum_v2_task.c +++ b/main/tasks/stratum_v2_task.c @@ -29,6 +29,11 @@ static const char *TAG = "stratum_v2_task"; +static void stratum_v2_free_extended_work(void *work) +{ + sv2_ext_job_free((sv2_ext_job_t *)work); +} + // Load authority pubkey from NVS (base58-encoded) into 32-byte buffer. // SV2 format: base58check(0x0001_LE + 32_byte_xonly_pubkey) // Decoded: 2-byte version + 32-byte pubkey + 4-byte checksum = 38 bytes @@ -90,6 +95,9 @@ static sv2_channel_type_t sv2_select_channel_type(GlobalState *GLOBAL_STATE, boo void stratum_v2_close_connection(GlobalState *GLOBAL_STATE) { ESP_LOGE(TAG, "Shutting down SV2 connection and restarting..."); + if (GLOBAL_STATE->sv2_conn) { + GLOBAL_STATE->sv2_conn->channel_opened = false; + } if (GLOBAL_STATE->sv2_noise_ctx) { sv2_noise_destroy(GLOBAL_STATE->sv2_noise_ctx); GLOBAL_STATE->sv2_noise_ctx = NULL; @@ -206,36 +214,45 @@ static void stratum_v2_enqueue_job(GlobalState *GLOBAL_STATE, sv2_conn_t *conn, SYSTEM_notify_new_ntime(GLOBAL_STATE, ntime); - if (clean_jobs && (GLOBAL_STATE->stratum_queue.count > 0)) { - SYSTEM_clean_jobs_queue(GLOBAL_STATE); + if (clean_jobs) { + SYSTEM_set_work_source(GLOBAL_STATE, WORK_ITEM_STRATUM_V2_STANDARD); } - if (GLOBAL_STATE->stratum_queue.count == QUEUE_SIZE) { - void *old = queue_dequeue(&GLOBAL_STATE->stratum_queue); - free(old); - } - - queue_enqueue(&GLOBAL_STATE->stratum_queue, job); + queue_enqueue(&GLOBAL_STATE->stratum_queue, + work_queue_item_create(&GLOBAL_STATE->stratum_queue, job, + WORK_ITEM_STRATUM_V2_STANDARD, free)); } // Enqueue an sv2_ext_job_t onto the stratum queue (extended channels) static void stratum_v2_enqueue_ext_job(GlobalState *GLOBAL_STATE, sv2_conn_t *conn, sv2_ext_job_t *job) { + if (conn->extranonce_prefix_len > sizeof(job->extranonce_prefix) || + conn->extranonce_size < 2 || + conn->extranonce_size > 32) { + ESP_LOGE(TAG, "Invalid extended-channel extranonce parameters"); + sv2_ext_job_free(job); + return; + } + + memcpy(job->extranonce_prefix, conn->extranonce_prefix, + conn->extranonce_prefix_len); + job->extranonce_prefix_len = conn->extranonce_prefix_len; + job->extranonce_size = conn->extranonce_size; + GLOBAL_STATE->SYSTEM_MODULE.work_received++; SYSTEM_notify_new_ntime(GLOBAL_STATE, job->ntime); - if (job->clean_jobs && (GLOBAL_STATE->stratum_queue.count > 0)) { - SYSTEM_clean_jobs_queue(GLOBAL_STATE); - } - - if (GLOBAL_STATE->stratum_queue.count == QUEUE_SIZE) { - void *old = queue_dequeue(&GLOBAL_STATE->stratum_queue); - sv2_ext_job_free((sv2_ext_job_t *)old); + if (job->clean_jobs) { + SYSTEM_set_work_source(GLOBAL_STATE, WORK_ITEM_STRATUM_V2_EXTENDED); } - queue_enqueue(&GLOBAL_STATE->stratum_queue, job); + queue_enqueue( + &GLOBAL_STATE->stratum_queue, + work_queue_item_create(&GLOBAL_STATE->stratum_queue, job, + WORK_ITEM_STRATUM_V2_EXTENDED, + stratum_v2_free_extended_work)); } // Decode coinbase from extended job prefix/suffix by converting to hex and reusing V1 decoder @@ -565,17 +582,10 @@ void stratum_v2_task(void *pvParameters) { GlobalState *GLOBAL_STATE = (GlobalState *)pvParameters; - // Determine channel type before setting up queue free function + // Determine channel type before opening the connection. bool use_fallback_init = GLOBAL_STATE->SYSTEM_MODULE.is_using_fallback; sv2_channel_type_t channel_type = sv2_select_channel_type(GLOBAL_STATE, use_fallback_init); - // Set V2-specific free function for the work queue - if (channel_type == SV2_CHANNEL_EXTENDED) { - GLOBAL_STATE->stratum_queue.free_fn = (void (*)(void *))sv2_ext_job_free; - } else { - GLOBAL_STATE->stratum_queue.free_fn = free; - } - // Set default version mask for version rolling GLOBAL_STATE->version_mask = STRATUM_DEFAULT_VERSION_MASK; GLOBAL_STATE->new_stratum_version_rolling_msg = true; @@ -910,6 +920,12 @@ void stratum_v2_task(void *pvParameters) conn->channel_opened = true; memcpy(conn->target, target, 32); + SYSTEM_set_work_source( + GLOBAL_STATE, + channel_type == SV2_CHANNEL_EXTENDED + ? WORK_ITEM_STRATUM_V2_EXTENDED + : WORK_ITEM_STRATUM_V2_STANDARD); + double pdiff = hash_to_pdiff(target); GLOBAL_STATE->pool_difficulty = pdiff; GLOBAL_STATE->new_set_mining_difficulty_msg = true; diff --git a/main/work_queue.c b/main/work_queue.c index 4ec6389e13..4db226893f 100644 --- a/main/work_queue.c +++ b/main/work_queue.c @@ -2,26 +2,104 @@ #include "esp_log.h" #include #include -#include void queue_init(work_queue *queue) { queue->head = 0; queue->tail = 0; queue->count = 0; - queue->free_fn = NULL; + queue->source_kind = WORK_ITEM_NONE; + queue->source_epoch = 1; pthread_mutex_init(&queue->lock, NULL); pthread_cond_init(&queue->not_empty, NULL); pthread_cond_init(&queue->not_full, NULL); } -void queue_enqueue(work_queue *queue, void *new_work) +work_queue_item_t work_queue_item_create(work_queue *queue, void *data, + work_item_kind_t kind, + void (*free_fn)(void *)) { + work_queue_item_t item = { + .data = data, + .kind = kind, + .free_fn = free_fn, + }; + pthread_mutex_lock(&queue->lock); + item.source_epoch = queue->source_epoch; + pthread_mutex_unlock(&queue->lock); + return item; +} - while (queue->count == QUEUE_SIZE) - { - pthread_cond_wait(&queue->not_full, &queue->lock); +void work_queue_item_free(work_queue_item_t *item) +{ + if (item == NULL) { + return; + } + + if (item->data != NULL) { + if (item->free_fn != NULL) { + item->free_fn(item->data); + } else { + free(item->data); + } + } + + *item = (work_queue_item_t) {0}; +} + +static void queue_clear_locked(work_queue *queue) +{ + while (queue->count > 0) { + work_queue_item_t next_work = queue->buffer[queue->head]; + work_queue_item_free(&next_work); + queue->head = (queue->head + 1) % QUEUE_SIZE; + queue->count--; + } +} + +void queue_set_source(work_queue *queue, work_item_kind_t kind) +{ + pthread_mutex_lock(&queue->lock); + queue_clear_locked(queue); + queue->source_kind = kind; + queue->source_epoch++; + if (queue->source_epoch == 0) { + queue->source_epoch = 1; + } + pthread_cond_broadcast(&queue->not_empty); + pthread_cond_broadcast(&queue->not_full); + pthread_mutex_unlock(&queue->lock); +} + +work_queue_source_t queue_get_source(work_queue *queue) +{ + pthread_mutex_lock(&queue->lock); + work_queue_source_t source = { + .kind = queue->source_kind, + .epoch = queue->source_epoch, + }; + pthread_mutex_unlock(&queue->lock); + return source; +} + +bool queue_enqueue(work_queue *queue, work_queue_item_t new_work) +{ + pthread_mutex_lock(&queue->lock); + + if (queue->source_kind == WORK_ITEM_NONE || + new_work.kind != queue->source_kind || + new_work.source_epoch != queue->source_epoch) { + pthread_mutex_unlock(&queue->lock); + work_queue_item_free(&new_work); + return false; + } + + if (queue->count == QUEUE_SIZE) { + work_queue_item_t old_work = queue->buffer[queue->head]; + work_queue_item_free(&old_work); + queue->head = (queue->head + 1) % QUEUE_SIZE; + queue->count--; } queue->buffer[queue->tail] = new_work; @@ -30,9 +108,10 @@ void queue_enqueue(work_queue *queue, void *new_work) pthread_cond_signal(&queue->not_empty); pthread_mutex_unlock(&queue->lock); + return true; } -void *queue_dequeue(work_queue *queue) +work_queue_item_t queue_dequeue(work_queue *queue) { pthread_mutex_lock(&queue->lock); @@ -41,7 +120,7 @@ void *queue_dequeue(work_queue *queue) pthread_cond_wait(&queue->not_empty, &queue->lock); } - void *next_work = queue->buffer[queue->head]; + work_queue_item_t next_work = queue->buffer[queue->head]; queue->head = (queue->head + 1) % QUEUE_SIZE; queue->count--; @@ -51,34 +130,34 @@ void *queue_dequeue(work_queue *queue) return next_work; } -void *queue_dequeue_timeout(work_queue *queue, int timeout_ms) +work_queue_item_t queue_dequeue_timeout(work_queue *queue, int timeout_ms) { pthread_mutex_lock(&queue->lock); - while (queue->count == 0) - { - struct timespec timeout_time; - clock_gettime(CLOCK_REALTIME, &timeout_time); - - // Add timeout_ms milliseconds to current time - timeout_time.tv_sec += timeout_ms / 1000; - timeout_time.tv_nsec += (timeout_ms % 1000) * 1000000; + if (queue->count == 0 && timeout_ms <= 0) { + pthread_mutex_unlock(&queue->lock); + return (work_queue_item_t) {0}; + } - // Handle nanosecond overflow - if (timeout_time.tv_nsec >= 1000000000) { - timeout_time.tv_sec += 1; - timeout_time.tv_nsec -= 1000000000; - } + uint32_t starting_epoch = queue->source_epoch; + struct timespec timeout_time; + clock_gettime(CLOCK_REALTIME, &timeout_time); + timeout_time.tv_sec += timeout_ms / 1000; + timeout_time.tv_nsec += (long)(timeout_ms % 1000) * 1000000L; + if (timeout_time.tv_nsec >= 1000000000L) { + timeout_time.tv_sec += 1; + timeout_time.tv_nsec -= 1000000000L; + } + while (queue->count == 0) { int result = pthread_cond_timedwait(&queue->not_empty, &queue->lock, &timeout_time); - if (result == ETIMEDOUT) { - // Timeout occurred, return NULL + if (queue->source_epoch != starting_epoch || result != 0) { pthread_mutex_unlock(&queue->lock); - return NULL; + return (work_queue_item_t) {0}; } } - void *next_work = queue->buffer[queue->head]; + work_queue_item_t next_work = queue->buffer[queue->head]; queue->head = (queue->head + 1) % QUEUE_SIZE; queue->count--; @@ -90,20 +169,5 @@ void *queue_dequeue_timeout(work_queue *queue, int timeout_ms) void queue_clear(work_queue *queue) { - pthread_mutex_lock(&queue->lock); - - while (queue->count > 0) - { - void *next_work = queue->buffer[queue->head]; - if (queue->free_fn) { - queue->free_fn(next_work); - } else { - free(next_work); - } - queue->head = (queue->head + 1) % QUEUE_SIZE; - queue->count--; - } - - pthread_cond_signal(&queue->not_full); - pthread_mutex_unlock(&queue->lock); + queue_set_source(queue, WORK_ITEM_NONE); } diff --git a/main/work_queue.h b/main/work_queue.h index 918df7fae6..1d8eaf4a8b 100644 --- a/main/work_queue.h +++ b/main/work_queue.h @@ -2,25 +2,56 @@ #define WORK_QUEUE_H #include +#include +#include #define QUEUE_SIZE 12 +typedef enum +{ + WORK_ITEM_NONE = 0, + WORK_ITEM_STRATUM_V1, + WORK_ITEM_STRATUM_V2_STANDARD, + WORK_ITEM_STRATUM_V2_EXTENDED, +} work_item_kind_t; + +typedef struct +{ + void *data; + work_item_kind_t kind; + uint32_t source_epoch; + void (*free_fn)(void *); +} work_queue_item_t; + +typedef struct +{ + work_item_kind_t kind; + uint32_t epoch; +} work_queue_source_t; + typedef struct { - void *buffer[QUEUE_SIZE]; + work_queue_item_t buffer[QUEUE_SIZE]; int head; int tail; int count; pthread_mutex_t lock; pthread_cond_t not_empty; pthread_cond_t not_full; - void (*free_fn)(void *); // Protocol-specific free function for queue items + work_item_kind_t source_kind; + uint32_t source_epoch; } work_queue; void queue_init(work_queue *queue); -void queue_enqueue(work_queue *queue, void *new_work); -void *queue_dequeue(work_queue *queue); -void *queue_dequeue_timeout(work_queue *queue, int timeout_ms); +work_queue_item_t work_queue_item_create(work_queue *queue, void *data, + work_item_kind_t kind, + void (*free_fn)(void *)); +void work_queue_item_free(work_queue_item_t *item); +void queue_set_source(work_queue *queue, work_item_kind_t kind); +work_queue_source_t queue_get_source(work_queue *queue); +bool queue_enqueue(work_queue *queue, work_queue_item_t new_work); +work_queue_item_t queue_dequeue(work_queue *queue); +work_queue_item_t queue_dequeue_timeout(work_queue *queue, int timeout_ms); void queue_clear(work_queue *queue); #endif // WORK_QUEUE_H diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index e98ced1503..eca113b5d7 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -3,14 +3,14 @@ cmake_minimum_required(VERSION 3.16) # Include the components directory of the main application: # -set(EXTRA_COMPONENT_DIRS "../components") +set(EXTRA_COMPONENT_DIRS "../components" "../test/components") # Set the components to include the tests for. # This can be overriden from CMake cache: # - when invoking CMake directly: cmake -D TEST_COMPONENTS="xxxxx" .. # - when using idf.py: idf.py -T xxxxx build # -set(TEST_COMPONENTS "stratum asic" CACHE STRING "List of components to test") +set(TEST_COMPONENTS "stratum asic work_queue_test" CACHE STRING "List of components to test") include($ENV{IDF_PATH}/tools/cmake/project.cmake) diff --git a/test/components/work_queue_test/CMakeLists.txt b/test/components/work_queue_test/CMakeLists.txt new file mode 100644 index 0000000000..aaeb8eb89a --- /dev/null +++ b/test/components/work_queue_test/CMakeLists.txt @@ -0,0 +1,2 @@ +idf_component_register(SRCS "../../../main/work_queue.c" + INCLUDE_DIRS "../../../main") diff --git a/test/components/work_queue_test/test/CMakeLists.txt b/test/components/work_queue_test/test/CMakeLists.txt new file mode 100644 index 0000000000..8b8a957f7c --- /dev/null +++ b/test/components/work_queue_test/test/CMakeLists.txt @@ -0,0 +1,3 @@ +idf_component_register(SRCS "test_work_queue.c" + INCLUDE_DIRS "." + REQUIRES unity work_queue_test) diff --git a/test/components/work_queue_test/test/test_work_queue.c b/test/components/work_queue_test/test/test_work_queue.c new file mode 100644 index 0000000000..ba33b318f5 --- /dev/null +++ b/test/components/work_queue_test/test/test_work_queue.c @@ -0,0 +1,103 @@ +#include + +#include "unity.h" +#include "work_queue.h" + +static int freed_v1; +static int freed_standard; +static int freed_extended; + +static void count_and_free(void *data, int *counter) +{ + TEST_ASSERT_NOT_NULL(data); + (*counter)++; + free(data); +} + +static void free_v1(void *data) +{ + count_and_free(data, &freed_v1); +} + +static void free_standard(void *data) +{ + count_and_free(data, &freed_standard); +} + +static void free_extended(void *data) +{ + count_and_free(data, &freed_extended); +} + +static work_queue_item_t make_item(work_queue *queue, work_item_kind_t kind, + void (*free_fn)(void *)) +{ + work_item_kind_t *stored_kind = malloc(sizeof(*stored_kind)); + TEST_ASSERT_NOT_NULL(stored_kind); + *stored_kind = kind; + return work_queue_item_create(queue, stored_kind, kind, free_fn); +} + +TEST_CASE("work queue binds ownership to source kind and epoch", "[work_queue]") +{ + work_queue queue; + queue_init(&queue); + freed_v1 = 0; + freed_standard = 0; + freed_extended = 0; + + queue_set_source(&queue, WORK_ITEM_STRATUM_V1); + work_queue_source_t v1_source = queue_get_source(&queue); + + TEST_ASSERT_TRUE(queue_enqueue( + &queue, make_item(&queue, WORK_ITEM_STRATUM_V1, free_v1))); + TEST_ASSERT_FALSE(queue_enqueue( + &queue, make_item(&queue, WORK_ITEM_STRATUM_V2_STANDARD, free_standard))); + + work_queue_item_t item = queue_dequeue(&queue); + TEST_ASSERT_EQUAL(WORK_ITEM_STRATUM_V1, item.kind); + TEST_ASSERT_EQUAL(WORK_ITEM_STRATUM_V1, + *(work_item_kind_t *)item.data); + TEST_ASSERT_EQUAL_UINT32(v1_source.epoch, item.source_epoch); + work_queue_item_free(&item); + + TEST_ASSERT_NULL(item.data); + TEST_ASSERT_EQUAL(WORK_ITEM_NONE, item.kind); + TEST_ASSERT_NULL(item.free_fn); + + work_queue_item_t stale = make_item(&queue, WORK_ITEM_STRATUM_V1, free_v1); + queue_set_source(&queue, WORK_ITEM_STRATUM_V1); + work_queue_source_t next_v1_source = queue_get_source(&queue); + TEST_ASSERT_NOT_EQUAL(v1_source.epoch, next_v1_source.epoch); + TEST_ASSERT_FALSE(queue_enqueue(&queue, stale)); + + queue_set_source(&queue, WORK_ITEM_STRATUM_V2_EXTENDED); + TEST_ASSERT_TRUE(queue_enqueue( + &queue, make_item(&queue, WORK_ITEM_STRATUM_V2_EXTENDED, free_extended))); + queue_clear(&queue); + TEST_ASSERT_EQUAL_INT(2, freed_v1); + TEST_ASSERT_EQUAL_INT(1, freed_standard); + TEST_ASSERT_EQUAL_INT(1, freed_extended); +} + +TEST_CASE("work queue replaces oldest item and timeout is empty", "[work_queue]") +{ + work_queue queue; + queue_init(&queue); + freed_v1 = 0; + queue_set_source(&queue, WORK_ITEM_STRATUM_V1); + + for (int i = 0; i < QUEUE_SIZE + 1; i++) { + TEST_ASSERT_TRUE(queue_enqueue( + &queue, make_item(&queue, WORK_ITEM_STRATUM_V1, free_v1))); + } + TEST_ASSERT_EQUAL_INT(1, freed_v1); + + queue_clear(&queue); + TEST_ASSERT_EQUAL_INT(QUEUE_SIZE + 1, freed_v1); + + work_queue_item_t item = queue_dequeue_timeout(&queue, 0); + TEST_ASSERT_NULL(item.data); + TEST_ASSERT_EQUAL(WORK_ITEM_NONE, item.kind); + TEST_ASSERT_NULL(item.free_fn); +}