Files
event_miner/event_miner.c

724 lines
24 KiB
C

/*
* Event Miner - Nostr Proof-of-Work Mining Tool
*
* A multithreaded command-line tool for adding NIP-13 Proof-of-Work to Nostr events.
* Uses the nostr_core_lib for cryptographic operations and event handling.
*/
#define _GNU_SOURCE // For strdup
#define _POSIX_C_SOURCE 200112L // For usleep
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <time.h>
#include <unistd.h>
#include <getopt.h>
#include <errno.h>
#include <signal.h>
#include <sys/types.h>
#include "nostr_core_lib/nostr_core/nostr_common.h" // Common definitions and init/cleanup
#include "nostr_core_lib/nostr_core/nip001.h" // Basic protocol functions
#include "nostr_core_lib/nostr_core/nip013.h" // Proof-of-work functions
#include "nostr_core_lib/nostr_core/nip019.h" // Bech32 decoding for nsec
#include "nostr_core_lib/cjson/cJSON.h"
// Constants
#define MAX_EVENT_SIZE 1048576 // 1MB max event size
#define DEFAULT_THREADS 4
#define DEFAULT_POW 2
// Thread exit codes
#define THREAD_EXIT_SUCCESS 0 // Found solution or normal completion
#define THREAD_EXIT_STOPPED 1 // Stopped by main thread
#define THREAD_EXIT_ERROR 2 // Error occurred
// Global variables for debugging
static volatile sig_atomic_t g_signal_received = 0;
static volatile int g_shutdown_requested = 0;
static pthread_t* g_thread_handles = NULL;
static int g_thread_count = 0;
static void* g_worker_contexts = NULL; // Will be cast to mining_context_t*
// Forward declarations for callbacks
typedef struct mining_context mining_context_t;
// Callback function types
typedef void (*solution_callback_t)(cJSON* solution, void* user_data);
// Main context for control decisions
typedef struct {
volatile int solution_found;
volatile int timeout_reached;
cJSON* result_event;
pthread_mutex_t result_mutex;
} main_context_t;
// Mining context for workers (keeping legacy fields for now during transition)
struct mining_context {
cJSON* event;
unsigned char private_key[32];
int target_difficulty;
int thread_id;
// Callbacks for reporting (no control decisions)
solution_callback_t solution_callback;
void* user_data;
// Control flag (only main thread modifies)
volatile int should_stop;
// Verbose mode and progress tracking
int verbose_enabled;
int best_leading_zeros;
time_t thread_start_time;
// Legacy fields for compatibility during transition
volatile int found;
cJSON* result_event;
pthread_mutex_t mutex;
pthread_cond_t cond;
time_t start_time;
int timeout_seconds;
int thread_count;
};
typedef struct {
int pow;
char* nsec;
int threads;
char* event_file;
int timeout_sec;
int verbose;
int help;
} args_t;
// Function declarations
static void usage(const char* prog_name);
static int parse_arguments(int argc, char* argv[], args_t* args);
static char* read_event_json(const char* filename);
static char* read_stdin_json(void);
static void* miner_thread(void* arg);
static int mine_event(mining_context_t* ctx);
static void cleanup_context(mining_context_t* ctx);
// Signal handling and debugging functions
static void signal_handler(int sig);
static void install_signal_handlers(void);
static void log_thread_exit(int thread_id, void* exit_status, const char* reason);
static const char* get_signal_name(int sig);
static void emergency_shutdown(void);
// Callback implementations
static void solution_found_callback(cJSON* solution, void* user_data);
static void verbose_pow_callback(int current_difficulty, uint64_t nonce, void* user_data);
// Usage information
static void usage(const char* prog_name) {
fprintf(stderr, "Usage: %s -pow <difficulty> -nsec <private_key> -threads <count> [options]\n\n", prog_name);
fprintf(stderr, "Required arguments:\n");
fprintf(stderr, " -pow <difficulty> Number of leading zero bits for proof-of-work\n");
fprintf(stderr, " -nsec <private_key> Private key in hex or nsec bech32 format\n");
fprintf(stderr, " -threads <count> Number of mining threads to use\n\n");
fprintf(stderr, "Optional arguments:\n");
fprintf(stderr, " -e <filename> Read event from file (default: stdin)\n");
fprintf(stderr, " --timeout_sec <sec> Timeout in seconds (default: no timeout)\n");
fprintf(stderr, " -v Verbose mode - show mining progress\n");
fprintf(stderr, " -h, --help Show this help message\n\n");
fprintf(stderr, "Examples:\n");
fprintf(stderr, " echo '{\"kind\":1,...}' | %s -pow 4 -nsec nsec1... -threads 8\n", prog_name);
fprintf(stderr, " %s -pow 4 -nsec abc123... -threads 8 -e event.json --timeout_sec 60\n", prog_name);
}
// Parse command line arguments
static int parse_arguments(int argc, char* argv[], args_t* args) {
// Initialize args structure
memset(args, 0, sizeof(args_t));
args->pow = -1; // Indicates not set
args->threads = -1; // Indicates not set
// Simple manual parsing to avoid getopt complexities with multi-char options
for (int i = 1; i < argc; i++) {
if (strcmp(argv[i], "-pow") == 0 && i + 1 < argc) {
args->pow = atoi(argv[i + 1]);
if (args->pow <= 0) {
fprintf(stderr, "Error: pow must be a positive integer\n");
return -1;
}
i++; // Skip the next argument
} else if (strcmp(argv[i], "-nsec") == 0 && i + 1 < argc) {
if (args->nsec) free(args->nsec);
args->nsec = strdup(argv[i + 1]);
if (!args->nsec) {
fprintf(stderr, "Error: memory allocation failed\n");
return -1;
}
i++; // Skip the next argument
} else if (strcmp(argv[i], "-threads") == 0 && i + 1 < argc) {
args->threads = atoi(argv[i + 1]);
if (args->threads <= 0) {
fprintf(stderr, "Error: threads must be a positive integer\n");
return -1;
}
i++; // Skip the next argument
} else if (strcmp(argv[i], "-e") == 0 && i + 1 < argc) {
args->event_file = strdup(argv[i + 1]);
if (!args->event_file) {
fprintf(stderr, "Error: memory allocation failed\n");
return -1;
}
i++; // Skip the next argument
} else if (strcmp(argv[i], "--timeout_sec") == 0 && i + 1 < argc) {
args->timeout_sec = atoi(argv[i + 1]);
if (args->timeout_sec <= 0) {
fprintf(stderr, "Error: timeout_sec must be a positive integer\n");
return -1;
}
i++; // Skip the next argument
} else if (strcmp(argv[i], "-v") == 0) {
args->verbose = 1;
} else if (strcmp(argv[i], "-h") == 0 || strcmp(argv[i], "--help") == 0) {
args->help = 1;
return 0;
} else if (argv[i][0] == '-') {
fprintf(stderr, "Error: Unknown option '%s'\n", argv[i]);
return -1;
}
}
// Check required arguments
if (args->pow == -1 || !args->nsec || args->threads == -1) {
fprintf(stderr, "Error: Missing required arguments\n");
return -1;
}
return 0;
}
// Read event JSON from file
static char* read_event_json(const char* filename) {
FILE* file = fopen(filename, "r");
if (!file) {
fprintf(stderr, "Error: Cannot open file '%s': %s\n", filename, strerror(errno));
return NULL;
}
// Get file size
fseek(file, 0, SEEK_END);
long file_size = ftell(file);
fseek(file, 0, SEEK_SET);
if (file_size > MAX_EVENT_SIZE) {
fprintf(stderr, "Error: Event file too large (max %d bytes)\n", MAX_EVENT_SIZE);
fclose(file);
return NULL;
}
// Allocate buffer and read file
char* buffer = malloc(file_size + 1);
if (!buffer) {
fprintf(stderr, "Error: Memory allocation failed\n");
fclose(file);
return NULL;
}
size_t bytes_read = fread(buffer, 1, file_size, file);
buffer[bytes_read] = '\0';
fclose(file);
return buffer;
}
// Read event JSON from stdin
static char* read_stdin_json(void) {
char* buffer = malloc(MAX_EVENT_SIZE);
if (!buffer) {
fprintf(stderr, "Error: Memory allocation failed\n");
return NULL;
}
size_t total_read = 0;
char chunk[4096];
while (fgets(chunk, sizeof(chunk), stdin)) {
size_t chunk_len = strlen(chunk);
if (total_read + chunk_len >= MAX_EVENT_SIZE - 1) {
fprintf(stderr, "Error: Input too large (max %d bytes)\n", MAX_EVENT_SIZE);
free(buffer);
return NULL;
}
strcpy(buffer + total_read, chunk);
total_read += chunk_len;
}
buffer[total_read] = '\0';
if (total_read == 0) {
fprintf(stderr, "Error: No input received\n");
free(buffer);
return NULL;
}
return buffer;
}
// Signal handling and debugging functions
static const char* get_signal_name(int sig) {
switch (sig) {
case SIGSEGV: return "SIGSEGV (Segmentation fault)";
case SIGABRT: return "SIGABRT (Abort)";
case SIGFPE: return "SIGFPE (Floating point exception)";
case SIGBUS: return "SIGBUS (Bus error)";
case SIGINT: return "SIGINT (Interrupt)";
case SIGTERM: return "SIGTERM (Terminate)";
default: return "Unknown signal";
}
}
static void log_thread_exit(int thread_id, void* exit_status, const char* reason) {
time_t current_time = time(NULL);
struct tm* local_time = localtime(&current_time);
printf("[%02d:%02d:%02d] Thread %d exited: %s (status: %ld)\n",
local_time->tm_hour, local_time->tm_min, local_time->tm_sec,
thread_id, reason, (long)exit_status);
fflush(stdout);
}
static void emergency_shutdown(void) {
// Set global shutdown flag
g_shutdown_requested = 1;
// Signal all worker threads to stop if contexts are available
if (g_worker_contexts) {
mining_context_t* contexts = (mining_context_t*)g_worker_contexts;
for (int i = 0; i < g_thread_count; i++) {
contexts[i].should_stop = 1;
}
}
}
static void signal_handler(int sig) {
static volatile sig_atomic_t in_signal_handler = 0;
// Prevent recursive signal handling
if (in_signal_handler) {
return;
}
in_signal_handler = 1;
g_signal_received = sig;
// Log the signal (async-signal-safe functions only)
const char* sig_name = get_signal_name(sig);
// Write to stderr using async-signal-safe functions
write(STDERR_FILENO, "\n[SIGNAL] Received ", 19);
write(STDERR_FILENO, sig_name, strlen(sig_name));
write(STDERR_FILENO, "\n", 1);
// For fatal signals, try emergency shutdown
if (sig == SIGSEGV || sig == SIGABRT || sig == SIGFPE || sig == SIGBUS) {
write(STDERR_FILENO, "[SIGNAL] Attempting emergency shutdown...\n", 42);
emergency_shutdown();
// Reset signal handler to default and re-raise
signal(sig, SIG_DFL);
raise(sig);
} else if (sig == SIGINT || sig == SIGTERM) {
// Graceful shutdown for interrupt signals
write(STDERR_FILENO, "[SIGNAL] Initiating graceful shutdown...\n", 41);
emergency_shutdown();
}
in_signal_handler = 0;
}
static void install_signal_handlers(void) {
// Install handlers for crash signals
signal(SIGSEGV, signal_handler);
signal(SIGABRT, signal_handler);
signal(SIGFPE, signal_handler);
signal(SIGBUS, signal_handler);
// Install handlers for graceful shutdown
signal(SIGINT, signal_handler);
signal(SIGTERM, signal_handler);
}
// Callback implementations
static void solution_found_callback(cJSON* solution, void* user_data) {
main_context_t* main_ctx = (main_context_t*)user_data;
pthread_mutex_lock(&main_ctx->result_mutex);
if (!main_ctx->solution_found) {
main_ctx->solution_found = 1;
main_ctx->result_event = cJSON_Duplicate(solution, 1);
}
pthread_mutex_unlock(&main_ctx->result_mutex);
}
// Verbose PoW callback - receives progress from nostr_add_proof_of_work
static void verbose_pow_callback(int current_difficulty, uint64_t nonce, void* user_data) {
mining_context_t* ctx = (mining_context_t*)user_data;
// Only report if verbose mode is enabled
if (!ctx->verbose_enabled) {
return;
}
// Update best difficulty achieved by this thread
if (current_difficulty > ctx->best_leading_zeros) {
ctx->best_leading_zeros = current_difficulty;
}
// Calculate mining rate (attempts per second)
time_t current_time = time(NULL);
time_t elapsed = current_time - ctx->thread_start_time;
double rate = elapsed > 0 ? (double)nonce / elapsed : 0.0;
// Format rate for display
char rate_str[32];
if (rate > 1000000) {
snprintf(rate_str, sizeof(rate_str), "%.1fM/sec", rate / 1000000.0);
} else if (rate > 1000) {
snprintf(rate_str, sizeof(rate_str), "%.1fk/sec", rate / 1000.0);
} else {
snprintf(rate_str, sizeof(rate_str), "%.0f/sec", rate);
}
// Print progress report
printf("[Thread %d] nonce: %llu, best: %d zeros, rate: %s, target: %d\n",
ctx->thread_id, (unsigned long long)nonce, ctx->best_leading_zeros,
rate_str, ctx->target_difficulty);
fflush(stdout);
}
// Mining thread function - Enhanced with exit status monitoring
static void* miner_thread(void* arg) {
mining_context_t* ctx = (mining_context_t*)arg;
void* exit_status = (void*)(intptr_t)THREAD_EXIT_ERROR; // Default to error
// Initialize thread-specific timing for verbose mode
ctx->thread_start_time = time(NULL);
ctx->best_leading_zeros = 0;
// Create a copy of the event for this thread
char* event_str = cJSON_Print(ctx->event);
if (!event_str) {
log_thread_exit(ctx->thread_id, exit_status, "JSON serialization failed");
return exit_status;
}
cJSON* local_event = cJSON_Parse(event_str);
free(event_str);
if (!local_event) {
log_thread_exit(ctx->thread_id, exit_status, "Event parsing failed");
return exit_status;
}
uint64_t attempts = 0;
// Mine until solution found or signaled to stop by main thread
while (!ctx->should_stop && !g_shutdown_requested) {
// Attempt mining with verbose callback if enabled
void (*progress_cb)(int, uint64_t, void*) = ctx->verbose_enabled ? verbose_pow_callback : NULL;
int result = nostr_add_proof_of_work(local_event, ctx->private_key,
ctx->target_difficulty, 1000000, // max_attempts
10000, // progress_report_interval
30, // timestamp_update_interval (seconds)
progress_cb, ctx);
attempts++;
if (result == NOSTR_SUCCESS) {
// Found solution - report to main thread via callback
if (ctx->solution_callback) {
ctx->solution_callback(local_event, ctx->user_data);
}
exit_status = (void*)(intptr_t)THREAD_EXIT_SUCCESS;
log_thread_exit(ctx->thread_id, exit_status, "Solution found");
break; // Exit after reporting solution
}
// Check for emergency shutdown
if (g_shutdown_requested) {
exit_status = (void*)(intptr_t)THREAD_EXIT_STOPPED;
log_thread_exit(ctx->thread_id, exit_status, "Emergency shutdown");
break;
}
// Small delay to prevent CPU overuse and allow responsive stopping
usleep(100); // 0.1ms - more responsive to should_stop signal
}
// Normal stop by main thread
if (ctx->should_stop && !g_shutdown_requested) {
exit_status = (void*)(intptr_t)THREAD_EXIT_STOPPED;
log_thread_exit(ctx->thread_id, exit_status, "Stopped by main thread");
}
cJSON_Delete(local_event);
return exit_status;
}
// Main mining function - Enhanced with debugging and monitoring
static int mine_event(mining_context_t* ctx) {
// Install signal handlers for crash detection
install_signal_handlers();
// Set up main context for centralized control
main_context_t main_ctx;
memset(&main_ctx, 0, sizeof(main_context_t));
// Initialize result mutex
if (pthread_mutex_init(&main_ctx.result_mutex, NULL) != 0) {
fprintf(stderr, "Error: Failed to initialize result mutex\n");
return -1;
}
// Set up callback system
ctx->solution_callback = solution_found_callback;
ctx->user_data = &main_ctx;
ctx->should_stop = 0;
// Create individual worker contexts (each gets its own thread_id)
mining_context_t* worker_contexts = malloc(ctx->thread_count * sizeof(mining_context_t));
if (!worker_contexts) {
fprintf(stderr, "Error: Memory allocation failed for worker contexts\n");
pthread_mutex_destroy(&main_ctx.result_mutex);
return -1;
}
// Copy base context to each worker and set thread_id
for (int i = 0; i < ctx->thread_count; i++) {
memcpy(&worker_contexts[i], ctx, sizeof(mining_context_t));
worker_contexts[i].thread_id = i;
}
// Create worker threads
pthread_t* threads = malloc(ctx->thread_count * sizeof(pthread_t));
if (!threads) {
fprintf(stderr, "Error: Memory allocation failed for threads\n");
free(worker_contexts);
pthread_mutex_destroy(&main_ctx.result_mutex);
return -1;
}
// Set up global debugging variables
g_thread_handles = threads;
g_thread_count = ctx->thread_count;
g_worker_contexts = worker_contexts;
time_t start_time = time(NULL);
printf("[DEBUG] Starting %d mining threads...\n", ctx->thread_count);
// Start threads
for (int i = 0; i < ctx->thread_count; i++) {
if (pthread_create(&threads[i], NULL, miner_thread, &worker_contexts[i]) != 0) {
fprintf(stderr, "Error: Failed to create thread %d\n", i);
// Stop already running threads
for (int j = 0; j < ctx->thread_count; j++) {
worker_contexts[j].should_stop = 1;
}
// Wait for threads that were created
for (int j = 0; j < i; j++) {
void* exit_status;
pthread_join(threads[j], &exit_status);
log_thread_exit(j, exit_status, "Cleanup after creation failure");
}
free(threads);
free(worker_contexts);
pthread_mutex_destroy(&main_ctx.result_mutex);
return -1;
}
printf("[DEBUG] Thread %d started successfully\n", i);
}
// Main thread control loop - centralized monitoring
int result = 0;
while (!main_ctx.solution_found && !main_ctx.timeout_reached && !g_shutdown_requested) {
// Check for timeout
if (ctx->timeout_seconds > 0) {
time_t current_time = time(NULL);
if (current_time - start_time >= ctx->timeout_seconds) {
main_ctx.timeout_reached = 1;
result = -1; // Timeout
break;
}
}
// Check for signals
if (g_signal_received) {
printf("[DEBUG] Signal received, shutting down...\n");
break;
}
// Small sleep to avoid busy waiting
usleep(10000); // 10ms
}
// Signal all workers to stop
printf("[DEBUG] Signaling threads to stop...\n");
for (int i = 0; i < ctx->thread_count; i++) {
worker_contexts[i].should_stop = 1;
}
// Wait for all threads to finish and capture exit statuses
printf("[DEBUG] Waiting for threads to finish...\n");
for (int i = 0; i < ctx->thread_count; i++) {
void* exit_status;
pthread_join(threads[i], &exit_status);
// Log exit status if it wasn't already logged by the thread
long status_code = (long)exit_status;
if (status_code == THREAD_EXIT_ERROR) {
log_thread_exit(i, exit_status, "Thread error (not previously logged)");
}
}
// Handle results
if (main_ctx.solution_found && main_ctx.result_event) {
ctx->result_event = main_ctx.result_event; // Transfer ownership
result = 1; // Success
printf("[DEBUG] Solution found successfully\n");
} else if (main_ctx.timeout_reached) {
result = -1; // Timeout
printf("[DEBUG] Mining timed out\n");
} else if (g_shutdown_requested || g_signal_received) {
result = -3; // Signal/emergency shutdown
printf("[DEBUG] Emergency shutdown completed\n");
} else {
result = -2; // Error
printf("[DEBUG] Mining failed with error\n");
}
// Clear global debugging variables
g_thread_handles = NULL;
g_thread_count = 0;
g_worker_contexts = NULL;
// Cleanup
free(threads);
free(worker_contexts);
pthread_mutex_destroy(&main_ctx.result_mutex);
return result;
}
// Cleanup context
static void cleanup_context(mining_context_t* ctx) {
if (ctx->event) {
cJSON_Delete(ctx->event);
ctx->event = NULL;
}
if (ctx->result_event) {
cJSON_Delete(ctx->result_event);
ctx->result_event = NULL;
}
}
// Main function
int main(int argc, char* argv[]) {
args_t args;
mining_context_t ctx;
int exit_code = 0;
// Initialize context
memset(&ctx, 0, sizeof(mining_context_t));
// Parse arguments
if (parse_arguments(argc, argv, &args) != 0) {
usage(argv[0]);
exit_code = 1;
goto cleanup_args;
}
if (args.help) {
usage(argv[0]);
goto cleanup_args;
}
// Initialize nostr library
if (nostr_init() != NOSTR_SUCCESS) {
fprintf(stderr, "Error: Failed to initialize nostr_core library\n");
exit_code = 1;
goto cleanup_args;
}
// Decode private key
if (nostr_decode_nsec(args.nsec, ctx.private_key) != NOSTR_SUCCESS) {
fprintf(stderr, "Error: Invalid private key format\n");
exit_code = 1;
goto cleanup_nostr;
}
// Read event JSON
char* event_json = NULL;
if (args.event_file) {
event_json = read_event_json(args.event_file);
} else {
event_json = read_stdin_json();
}
if (!event_json) {
exit_code = 1;
goto cleanup_nostr;
}
// Parse JSON event
ctx.event = cJSON_Parse(event_json);
free(event_json);
if (!ctx.event) {
fprintf(stderr, "Error: Invalid JSON event format\n");
exit_code = 1;
goto cleanup_nostr;
}
// Set mining parameters
ctx.target_difficulty = args.pow;
ctx.thread_count = args.threads;
ctx.timeout_seconds = args.timeout_sec > 0 ? args.timeout_sec : 0;
ctx.verbose_enabled = args.verbose;
// Start mining
int mining_result = mine_event(&ctx);
if (mining_result == 1 && ctx.result_event) {
// Success - output mined event
char* output_json = cJSON_Print(ctx.result_event);
if (output_json) {
printf("%s\n", output_json);
free(output_json);
} else {
fprintf(stderr, "Error: Failed to serialize result event\n");
exit_code = 1;
}
} else if (mining_result == -1) {
printf("timeout\n");
exit_code = 1;
} else {
fprintf(stderr, "Error: Mining failed\n");
exit_code = 1;
}
// Cleanup
cleanup_context(&ctx);
cleanup_nostr:
nostr_cleanup();
cleanup_args:
if (args.nsec) free(args.nsec);
if (args.event_file) free(args.event_file);
return exit_code;
}