/*

* Copyright 2014-2020,2023 Con Kolivas

*

* This program is free software; you can redistribute it and/or modify it

* under the terms of the GNU General Public License as published by the Free

* Software Foundation; either version 3 of the License, or (at your option)

* any later version. See COPYING for more details.

*/

#include "config.h"

#include <arpa/inet.h>

#include <sys/socket.h>

#include <sys/stat.h>

#include <sys/time.h>

#include <sys/types.h>

#include <ctype.h>

#include <dirent.h>

#include <fcntl.h>

#include <math.h>

#include <string.h>

#include <unistd.h>

#ifdef HAVE_ZMQ_H

#include <zmq.h>

#endif

#include "ckpool.h"

#include "libckpool.h"

#include "bitcoin.h"

#include "sha2.h"

#include "stratifier.h"

#include "uthash.h"

#include "utlist.h"

#include "connector.h"

#include "generator.h"

/* Consistent across all pool instances */

static const char *workpadding = "000000800000000000000000000000000000000000000000000000000000000000000000000000000000000080020000";

static const char *scriptsig_header = "01000000010000000000000000000000000000000000000000000000000000000000000000ffffffff";

static uchar scriptsig_header_bin[41];

static const double nonces = 4294967296;

/* Add unaccounted shares when they arrive, remove them with each update of

* rolling stats. */

struct pool_stats {

tv_t start_time;

ts_t last_update;

int workers;

int users;

int disconnected;

int remote_workers;

int remote_users;

/* Absolute shares stats */

int64_t unaccounted_shares;

int64_t accounted_shares;

/* Cycle of 32 to determine which users to dump stats on */

uint8_t userstats_cycle;

/* Shares per second for 1/5/15/60 minute rolling averages */

double sps1;

double sps5;

double sps15;

double sps60;

/* Diff shares stats */

int64_t unaccounted_diff_shares;

int64_t accounted_diff_shares;

int64_t unaccounted_rejects;

int64_t accounted_rejects;

/* Diff shares per second for 1/5/15... minute rolling averages */

double dsps1;

double dsps5;

double dsps15;

double dsps60;

double dsps360;

double dsps1440;

double dsps10080;

double network_diff;

int64_t best_diff;

};

typedef struct pool_stats pool_stats_t;

typedef struct genwork workbase_t;

struct json_params {

json_t *method;

json_t *params;

json_t *id_val;

int64_t client_id;

};

typedef struct json_params json_params_t;

/* Stratum json messages with their associated client id */

struct smsg {

json_t *json_msg;

int64_t client_id;

};

typedef struct smsg smsg_t;

struct userwb {

UT_hash_handle hh;

int64_t id;

uchar *coinb2bin; // Coinb2 cointaining this user's address for generation

char *coinb2;

int coinb2len; // Length of user coinb2

};

struct user_instance;

struct worker_instance;

struct stratum_instance;

typedef struct user_instance user_instance_t;

typedef struct worker_instance worker_instance_t;

typedef struct stratum_instance stratum_instance_t;

struct user_instance {

UT_hash_handle hh;

char username[128];

int id;

char *secondaryuserid;

bool btcaddress;

bool script;

/* Set for a solo user whose username did not classify as a BCH

* address: authorised anyway (D2 smart-fallback decision), mining to

* the pool's own btcaddress via a copy of sdata->txnbin/txnlen. */

bool pool_fallback;

/* A linked list of all connected instances of this user */

stratum_instance_t *clients;

/* A linked list of all connected workers of this user */

worker_instance_t *worker_instances;

int workers;

int remote_workers;

char txnbin[48];

int txnlen;

struct userwb *userwbs; /* Protected by instance lock */

double best_diff; /* Best share found by this user */

int64_t best_ever; /* Best share ever found by this user */

int64_t shares;

int64_t uadiff; /* Shares not yet accounted for in hashmeter */

double dsps1; /* Diff shares per second, 1 minute rolling average */

double dsps5; /* ... 5 minute ... */

double dsps60;/* etc */

double dsps1440;

double dsps10080;

tv_t last_share;

tv_t last_decay;

bool authorised; /* Has this username ever been authorised? */

time_t auth_time;

time_t failed_authtime; /* Last time this username failed to authorise */

int auth_backoff; /* How long to reject any auth attempts since last failure */

bool throttled; /* Have we begun rejecting auth attempts */

};

/* Combined data from workers with the same workername */

struct worker_instance {

user_instance_t *user_instance;

char *workername;

/* Number of stratum instances attached as this one worker */

int instance_count;

worker_instance_t *next;

worker_instance_t *prev;

int64_t shares;

int64_t uadiff; /* Shares not yet accounted for in hashmeter */

double dsps1;

double dsps5;

double dsps60;

double dsps1440;

double dsps10080;

tv_t last_share;

tv_t last_decay;

time_t start_time;

double best_diff; /* Best share found by this worker */

int64_t best_ever; /* Best share ever found by this worker */

int mindiff; /* User chosen mindiff */

bool idle;

bool notified_idle;

};

typedef struct stratifier_data sdata_t;

typedef struct proxy_base proxy_t;

/* Per client stratum instance == workers */

struct stratum_instance {

UT_hash_handle hh;

int64_t id;

/* Virtualid used as unique local id for passthrough clients */

int64_t virtualid;

stratum_instance_t *recycled_next;

stratum_instance_t *recycled_prev;

stratum_instance_t *user_next;

stratum_instance_t *user_prev;

stratum_instance_t *node_next;

stratum_instance_t *node_prev;

stratum_instance_t *remote_next;

stratum_instance_t *remote_prev;

/* Descriptive of ID number and passthrough if any */

char identity[128];

/* Reference count for when this instance is used outside of the

* instance_lock */

int ref;

char enonce1[36]; /* Fit up to 16 byte binary enonce1 */

uchar enonce1bin[16];

char enonce1var[20]; /* Fit up to 8 byte binary enonce1var */

uint64_t enonce1_64;

int session_id;

int64_t diff; /* Current diff */

int64_t old_diff; /* Previous diff */

int64_t diff_change_job_id; /* Last job_id we changed diff */

int64_t uadiff; /* Shares not yet accounted for in hashmeter */

double dsps1; /* Diff shares per second, 1 minute rolling average */

double dsps5; /* ... 5 minute ... */

double dsps60;/* etc */

double dsps1440;

double dsps10080;

tv_t ldc; /* Last diff change */

int ssdc; /* Shares since diff change */

tv_t first_share;

tv_t last_share;

tv_t last_decay;

time_t first_invalid; /* Time of first invalid in run of non stale rejects */

time_t upstream_invalid; /* As first_invalid but for upstream responses */

time_t start_time;

char address[INET6_ADDRSTRLEN];

bool node; /* Is this a mining node */

bool subscribed;

bool authorising; /* In progress, protected by instance_lock */

bool authorised;

bool dropped;

bool idle;

int reject; /* Indicator that this client is having a run of rejects

* or other problem and should be dropped lazily if

* this is set to 2 */

int latency; /* Latency when on a mining node */

bool reconnect; /* This client really needs to reconnect */

time_t reconnect_request; /* The time we sent a reconnect message */

user_instance_t *user_instance;

worker_instance_t *worker_instance;

char *useragent;

char *workername;

char *password;

bool messages; /* Is this a client that understands stratum messages */

int user_id;

int server; /* Which server is this instance bound to */

ckpool_t *ckp;

time_t last_txns; /* Last time this worker requested txn hashes */

time_t disconnected_time; /* Time this instance disconnected */

int64_t suggest_diff; /* Stratum client suggested diff */

double best_diff; /* Best share found by this instance */

sdata_t *sdata; /* Which sdata this client is bound to */

proxy_t *proxy; /* Proxy this is bound to in proxy mode */

int proxyid; /* Which proxy id */

int subproxyid; /* Which subproxy */

bool passthrough; /* Is this a passthrough */

bool trusted; /* Is this a trusted remote server */

bool remote; /* Is this a remote client on a trusted remote server */

bool rental_diff; /* Difficulty already applied from useragent-based rental

* service detection (NiceHash/MiningRigRentals), so the

* mindiff_overrides pattern loop must not clobber it */

};

struct share {

UT_hash_handle hh;

uchar hash[32];

int64_t workbase_id;

};

typedef struct share share_t;

struct proxy_base {

UT_hash_handle hh;

UT_hash_handle sh; /* For subproxy hashlist */

proxy_t *next; /* For retired subproxies */

proxy_t *prev;

int id;

int subid;

/* Priority has the user id encoded in the high bits if it's not a

* global proxy. */

int64_t priority;

bool global; /* Is this a global proxy */

int userid; /* Userid for non global proxies */

double diff;

char baseurl[128];

char url[128];

char auth[128];

char pass[128];

char enonce1[32];

uchar enonce1bin[16];

int enonce1constlen;

int enonce1varlen;

int nonce2len;

int enonce2varlen;

bool subscribed;

bool notified;

int64_t clients; /* Incrementing client count */

int64_t max_clients; /* Maximum number of clients per subproxy */

int64_t bound_clients; /* Currently actively bound clients */

int64_t combined_clients; /* Total clients of all subproxies of a parent proxy */

int64_t headroom; /* Temporary variable when calculating how many more clients can bind */

int subproxy_count; /* Number of subproxies */

proxy_t *parent; /* Parent proxy of each subproxy */

proxy_t *subproxies; /* Hashlist of subproxies sorted by subid */

sdata_t *sdata; /* Unique stratifer data for each subproxy */

bool dead;

bool deleted;

};

typedef struct session session_t;

struct session {

UT_hash_handle hh;

int session_id;

uint64_t enonce1_64;

int64_t client_id;

int userid;

time_t added;

char address[INET6_ADDRSTRLEN];

};

typedef struct txntable txntable_t;

struct txntable {

UT_hash_handle hh;

int id;

char hash[68];

char *data;

int refcount;

bool seen;

};

#define ID_AUTH 0

#define ID_WORKINFO 1

#define ID_AGEWORKINFO 2

#define ID_SHARES 3

#define ID_SHAREERR 4

#define ID_POOLSTATS 5

#define ID_WORKERSTATS 6

#define ID_BLOCK 7

#define ID_ADDRAUTH 8

#define ID_HEARTBEAT 9

struct stratifier_data {

ckpool_t *ckp;

char txnbin[48];

int txnlen;

char pooltxnbin[48];

int pooltxnlen;

pool_stats_t stats;

/* Protects changes to pool stats */

mutex_t stats_lock;

/* Protects changes to unaccounted pool stats */

mutex_t uastats_lock;

bool verbose;

uint64_t enonce1_64;

/* For protecting the txntable data */

cklock_t txn_lock;

/* For protecting the hashtable data */

cklock_t workbase_lock;

/* For the hashtable of all workbases */

workbase_t *workbases;

workbase_t *current_workbase;

int workbases_generated;

txntable_t *txns;

int64_t txns_generated;

/* Workbases from remote trusted servers */

workbase_t *remote_workbases;

/* Is this a node and unable to rebuild workinfos due to lack of txns */

bool wbincomplete;

/* Semaphore to serialise calls to add_base */

sem_t update_sem;

/* Time we last sent out a stratum update */

time_t update_time;

int64_t workbase_id;

int64_t blockchange_id;

int session_id;

char lasthash[68];

char lastswaphash[68];

ckmsgq_t *updateq; // Generator base work updates

ckmsgq_t *ssends; // Stratum sends

ckmsgq_t *srecvs; // Stratum receives

ckmsgq_t *sshareq; // Stratum share sends

ckmsgq_t *sauthq; // Stratum authorisations

ckmsgq_t *stxnq; // Transaction requests

int user_instance_id;

stratum_instance_t *stratum_instances;

stratum_instance_t *recycled_instances;

stratum_instance_t *node_instances;

stratum_instance_t *remote_instances;

int64_t stratum_generated;

int64_t disconnected_generated;

int64_t userwbs_generated;

session_t *disconnected_sessions;

user_instance_t *user_instances;

/* Protects both stratum and user instances */

cklock_t instance_lock;

share_t *shares;

mutex_t share_lock;

int64_t shares_generated;

int proxy_count; /* Total proxies generated (not necessarily still alive) */

proxy_t *proxy; /* Current proxy in use */

proxy_t *proxies; /* Hashlist of all proxies */

mutex_t proxy_lock; /* Protects all proxy data */

proxy_t *subproxy; /* Which subproxy this sdata belongs to in proxy mode */

};

typedef struct json_entry json_entry_t;

struct json_entry {

json_entry_t *next;

json_entry_t *prev;

json_t *val;

};

/* Priority levels for generator messages */

#define GEN_LAX 0

#define GEN_NORMAL 1

#define GEN_PRIORITY 2

/* For storing a set of messages within another lock, allowing us to dump them

* to the log outside of lock */

static void add_msg_entry(char_entry_t **entries, char **buf)

{

char_entry_t *entry;

if (!*buf)

return;

entry = ckalloc(sizeof(char_entry_t));

entry->buf = *buf;

*buf = NULL;

DL_APPEND(*entries, entry);

}

static void notice_msg_entries(char_entry_t **entries)

{

char_entry_t *entry, *tmpentry;

DL_FOREACH_SAFE(*entries, entry, tmpentry) {

DL_DELETE(*entries, entry);

LOGNOTICE("%s", entry->buf);

free(entry->buf);

free(entry);

}

}

static void info_msg_entries(char_entry_t **entries)

{

char_entry_t *entry, *tmpentry;

DL_FOREACH_SAFE(*entries, entry, tmpentry) {

DL_DELETE(*entries, entry);

LOGINFO("%s", entry->buf);

free(entry->buf);

free(entry);

}

}

static const int witnessdata_size = 36; // commitment header + hash

static void generate_coinbase(ckpool_t *ckp, workbase_t *wb)

{

uint64_t *u64, g64, d64 = 0;

sdata_t *sdata = ckp->sdata;

bool poolfee_dust = false;

char header[272];

int len, ofs = 0;

ts_t now;

/* Set fixed length coinb1 arrays to be more than enough */

wb->coinb1 = ckzalloc(256);

wb->coinb1bin = ckzalloc(128);

/* Strings in wb should have been zero memset prior. Generate binary

* templates first, then convert to hex */

memcpy(wb->coinb1bin, scriptsig_header_bin, 41);

ofs += 41; // Fixed header length;

ofs++; // Script length is filled in at the end @wb->coinb1bin[41];

/* Put block height at start of template */

len = ser_number(wb->coinb1bin + ofs, wb->height);

ofs += len;

/* Followed by flag */

len = strlen(wb->flags) / 2;

wb->coinb1bin[ofs++] = len;

hex2bin(wb->coinb1bin + ofs, wb->flags, len);

ofs += len;

/* Followed by timestamp */

ts_realtime(&now);

len = ser_number(wb->coinb1bin + ofs, now.tv_sec);

ofs += len;

/* Followed by our unique randomiser based on the nsec timestamp */

len = ser_number(wb->coinb1bin + ofs, now.tv_nsec);

ofs += len;

wb->enonce1varlen = ckp->nonce1length;

wb->enonce2varlen = ckp->nonce2length;

wb->coinb1bin[ofs++] = wb->enonce1varlen + wb->enonce2varlen;

wb->coinb1len = ofs;

len = wb->coinb1len - 41;

len += wb->enonce1varlen;

len += wb->enonce2varlen;

wb->coinb2bin = ckzalloc(512);

wb->coinb2len = 0; /* Start with empty coinbase, use only btcsig */

if (ckp->btcsig) {

int siglen = strlen(ckp->btcsig);

LOGDEBUG("Len %d sig %s", siglen, ckp->btcsig);

if (siglen) {

wb->coinb2bin[wb->coinb2len++] = siglen;

memcpy(wb->coinb2bin + wb->coinb2len, ckp->btcsig, siglen);

wb->coinb2len += siglen;

}

}

len += wb->coinb2len;

wb->coinb1bin[41] = len - 1; /* Set the length now */

__bin2hex(wb->coinb1, wb->coinb1bin, wb->coinb1len);

LOGDEBUG("Coinb1: %s", wb->coinb1);

/* Coinbase 1 complete */

memcpy(wb->coinb2bin + wb->coinb2len, "\xff\xff\xff\xff", 4);

wb->coinb2len += 4;

// Generation value

g64 = wb->coinbasevalue;

/* Check for pool operator fee or donation */

if (ckp->poolvalid && ckp->poolfee > 0) {

/* Pool operator fee takes precedence */

double dbl64 = (double)g64 / 100 * ckp->poolfee;

d64 = dbl64;

/* Dust guard: a fee output below the 546 sat dust threshold

* is non-standard and may be rejected by the network. Fall

* through to the single-output path for this workbase,

* paying the miner the full coinbasevalue, rather than

* emitting a dust output. Outputs still sum exactly to

* wb->coinbasevalue either way. */

if (d64 < 546) {

LOGWARNING("Pool fee output %"PRIu64" sats below dust threshold, omitting fee output", d64);

d64 = 0;

g64 = wb->coinbasevalue;

wb->coinb2bin[wb->coinb2len++] = 1 + wb->insert_witness;

poolfee_dust = true;

} else {

g64 -= d64; // To guarantee integers add up to the original coinbasevalue

wb->coinb2bin[wb->coinb2len++] = 2 + wb->insert_witness;

}

} else

wb->coinb2bin[wb->coinb2len++] = 1 + wb->insert_witness;

u64 = (uint64_t *)&wb->coinb2bin[wb->coinb2len];

*u64 = htole64(g64);

wb->coinb2len += 8;

/* Coinb2 address goes here, takes up 23~25 bytes + 1 byte for length */

wb->coinb3len = 0;

wb->coinb3bin = ckzalloc(256 + wb->insert_witness * (8 + witnessdata_size + 2));

/* Add pool operator fee or donation output */

if (ckp->poolvalid && ckp->poolfee > 0 && !poolfee_dust) {

u64 = (uint64_t *)wb->coinb3bin;

*u64 = htole64(d64);

wb->coinb3len += 8;

wb->coinb3bin[wb->coinb3len++] = sdata->pooltxnlen;

memcpy(wb->coinb3bin + wb->coinb3len, sdata->pooltxnbin, sdata->pooltxnlen);

wb->coinb3len += sdata->pooltxnlen;

} else if (!poolfee_dust) {

/* Only permanently disable poolfee when it isn't configured

* at all. A dust-suppressed poolfee is a per-workbase

* condition (coinbasevalue too small this round) and must

* not disable future, non-dust workbases. */

ckp->poolfee = 0;

}

if (wb->insert_witness) {

// 0 value

wb->coinb3len += 8;

wb->coinb3bin[wb->coinb3len++] = witnessdata_size + 2; // total scriptPubKey size

wb->coinb3bin[wb->coinb3len++] = 0x6a; // OP_RETURN

wb->coinb3bin[wb->coinb3len++] = witnessdata_size;

hex2bin(&wb->coinb3bin[wb->coinb3len], wb->witnessdata, witnessdata_size);

wb->coinb3len += witnessdata_size;

}

wb->coinb3len += 4; // Blank lock

if (!ckp->btcsolo) {

int coinbase_len, offset = 0;

char *coinbase, *cb;

json_t *val = NULL;

/* Append the generation address and coinb3 in !solo mode */

wb->coinb2bin[wb->coinb2len++] = sdata->txnlen;

memcpy(wb->coinb2bin + wb->coinb2len, sdata->txnbin, sdata->txnlen);

wb->coinb2len += sdata->txnlen;

memcpy(wb->coinb2bin + wb->coinb2len, wb->coinb3bin, wb->coinb3len);

wb->coinb2len += wb->coinb3len;

wb->coinb3len = 0;

dealloc(wb->coinb3bin);

/* Set this only once */

if (unlikely(!ckp->coinbase_valid)) {

/* We have enough to test the validity of the coinbase here */

coinbase_len = wb->coinb1len + ckp->nonce1length + ckp->nonce2length + wb->coinb2len;

coinbase = ckzalloc(coinbase_len);

memcpy(coinbase, wb->coinb1bin, wb->coinb1len);

offset += wb->coinb1len;

/* Space for nonce1 and 2 */

offset += ckp->nonce1length + ckp->nonce2length;

memcpy(coinbase + offset, wb->coinb2bin, wb->coinb2len);

offset += wb->coinb2len;

cb = bin2hex(coinbase, offset);

LOGDEBUG("Coinbase txn %s", cb);

free(coinbase);

if (generator_checktxn(ckp, cb, &val)) {

char *s = json_dumps(val, JSON_NO_UTF8 | JSON_COMPACT);

json_decref(val);

LOGNOTICE("Coinbase transaction confirmed valid");

LOGDEBUG("%s", s);

free(s);

} else {

/* This is a fatal error */

LOGEMERG("Coinbase failed valid transaction check, aborting!");

exit(1);

}

free(cb);

ckp->coinbase_valid = true;

LOGWARNING("Mining from any incoming username to address %s", ckp->btcaddress);

if (ckp->poolfee > 0 && ckp->poolvalid)

LOGWARNING("%.1f percent pool operator fee to %s", ckp->poolfee, ckp->pooladdress);

}

} else if (unlikely(!ckp->coinbase_valid)) {

/* Create a sample coinbase to test its validity in solo mode */

int coinbase_len, offset = 0;

char *coinbase, *cb;

json_t *val = NULL;

coinbase_len = wb->coinb1len + ckp->nonce1length + ckp->nonce2length + wb->coinb2len +

sdata->txnlen + wb->coinb3len + 1;

coinbase = ckzalloc(coinbase_len);

memcpy(coinbase, wb->coinb1bin, wb->coinb1len);

offset += wb->coinb1len;

offset += ckp->nonce1length + ckp->nonce2length;

memcpy(coinbase + offset, wb->coinb2bin, wb->coinb2len);

offset += wb->coinb2len;

coinbase[offset] = sdata->txnlen;

offset += 1;

memcpy(coinbase + offset, sdata->txnbin, sdata->txnlen);

offset += sdata->txnlen;

memcpy(coinbase + offset, wb->coinb3bin, wb->coinb3len);

offset += wb->coinb3len;

cb = bin2hex(coinbase, offset);

LOGDEBUG("Coinbase txn %s", cb);

free(coinbase);

if (generator_checktxn(ckp, cb, &val)) {

char *s = json_dumps(val, JSON_NO_UTF8 | JSON_COMPACT);

json_decref(val);

LOGNOTICE("Coinbase transaction confirmed valid");

LOGDEBUG("%s", s);

free(s);

} else {

/* This is a fatal error */

LOGEMERG("Coinbase failed valid transaction check, aborting!");

exit(1);

}

free(cb);

ckp->coinbase_valid = true;

LOGWARNING("Mining solo to any incoming valid BTC address username");

}

/* Set this just for node compatibility, though it's unused */

wb->coinb2 = bin2hex(wb->coinb2bin, wb->coinb2len);

LOGDEBUG("Coinb2: %s", wb->coinb2);

/* Coinbases 2 +/- 3 templates complete */

snprintf(header, 270, "%s%s%s%s%s%s%s",

wb->bbversion, wb->prevhash,

"0000000000000000000000000000000000000000000000000000000000000000",

wb->ntime, wb->nbit,

"00000000", /* nonce */

workpadding);

header[224] = 0;

LOGDEBUG("Header: %s", header);

hex2bin(wb->headerbin, header, 112);

}

static void stratum_broadcast_update(sdata_t *sdata, const workbase_t *wb, bool clean);

static void stratum_broadcast_updates(sdata_t *sdata, bool clean);

static void clear_userwb(sdata_t *sdata, int64_t id)

{

user_instance_t *instance, *tmp;

ck_wlock(&sdata->instance_lock);

HASH_ITER(hh, sdata->user_instances, instance, tmp) {

struct userwb *userwb;

HASH_FIND_I64(instance->userwbs, &id, userwb);

if (!userwb)

continue;

HASH_DEL(instance->userwbs, userwb);

free(userwb->coinb2bin);

free(userwb->coinb2);

free(userwb);

}

ck_wunlock(&sdata->instance_lock);

}

static void clear_workbase(ckpool_t *ckp, workbase_t *wb)

{

if (ckp->btcsolo)

clear_userwb(ckp->sdata, wb->id);

free(wb->flags);

free(wb->txn_data);

free(wb->txn_hashes);

free(wb->logdir);

free(wb->coinb1bin);

free(wb->coinb1);

free(wb->coinb2bin);

free(wb->coinb2);

free(wb->coinb3bin);

json_decref(wb->merkle_array);

if (wb->json)

json_decref(wb->json);

free(wb);

}

/* Remove all shares with a workbase id less than wb_id for block changes */

static void purge_share_hashtable(sdata_t *sdata, const int64_t wb_id)

{

share_t *share, *tmp;

int purged = 0;

mutex_lock(&sdata->share_lock);

HASH_ITER(hh, sdata->shares, share, tmp) {

if (share->workbase_id < wb_id) {

HASH_DEL(sdata->shares, share);

dealloc(share);

purged++;

}

}

mutex_unlock(&sdata->share_lock);

if (purged)

LOGINFO("Cleared %d shares from share hashtable", purged);

}

/* Remove all shares with a workbase id == wb_id being discarded */

static void age_share_hashtable(sdata_t *sdata, const int64_t wb_id)

{

share_t *share, *tmp;

int aged = 0;

mutex_lock(&sdata->share_lock);

HASH_ITER(hh, sdata->shares, share, tmp) {

if (share->workbase_id == wb_id) {

HASH_DEL(sdata->shares, share);

dealloc(share);

aged++;

}

}

mutex_unlock(&sdata->share_lock);

if (aged)

LOGINFO("Aged %d shares from share hashtable", aged);

}

/* Append a bulk list already created to the ssends list */

static void ssend_bulk_append(sdata_t *sdata, ckmsg_t *bulk_send, const int messages)

{

ckmsgq_t *ssends = sdata->ssends;

mutex_lock(ssends->lock);

ssends->messages += messages;

DL_CONCAT(ssends->msgs, bulk_send);

pthread_cond_signal(ssends->cond);

mutex_unlock(ssends->lock);

}

/* As ssend_bulk_append but for high priority messages to be put at the front

* of the list. */

static void ssend_bulk_prepend(sdata_t *sdata, ckmsg_t *bulk_send, const int messages)

{

ckmsgq_t *ssends = sdata->ssends;

ckmsg_t *tmp;

mutex_lock(ssends->lock);

tmp = ssends->msgs;

ssends->msgs = bulk_send;

ssends->messages += messages;

DL_CONCAT(ssends->msgs, tmp);

pthread_cond_signal(ssends->cond);

mutex_unlock(ssends->lock);

}

/* Send a json msg to an upstream trusted remote server */

static void upstream_json(ckpool_t *ckp, json_t *val)

{

char *msg;

msg = json_dumps(val, JSON_NO_UTF8 | JSON_PRESERVE_ORDER | JSON_COMPACT | JSON_EOL);

/* Connector absorbs and frees msg */

connector_upstream_msg(ckp, msg);

}

/* Upstream a json msgtype */

static void upstream_json_msgtype(ckpool_t *ckp, json_t *val, const int msg_type)

{

json_set_string(val, "method", stratum_msgs[msg_type]);

upstream_json(ckp, val);

}

/* Upstream a json msgtype, duplicating the json */

static void upstream_msgtype(ckpool_t *ckp, const json_t *val, const int msg_type)

{

json_t *json_msg = json_deep_copy(val);

json_set_string(json_msg, "method", stratum_msgs[msg_type]);

upstream_json(ckp, json_msg);

json_decref(json_msg);

}

static void send_node_workinfo(ckpool_t *ckp, sdata_t *sdata, const workbase_t *wb)

{

stratum_instance_t *client;

ckmsg_t *bulk_send = NULL;

int messages = 0;

json_t *wb_val;

wb_val = json_object();

json_set_int(wb_val, "jobid", wb->mapped_id);

json_set_string(wb_val, "target", wb->target);

json_set_double(wb_val, "diff", wb->diff);

json_set_int(wb_val, "version", wb->version);

json_set_int(wb_val, "curtime", wb->curtime);

json_set_string(wb_val, "prevhash", wb->prevhash);

json_set_string(wb_val, "ntime", wb->ntime);

json_set_string(wb_val, "bbversion", wb->bbversion);

json_set_string(wb_val, "nbit", wb->nbit);

json_set_int(wb_val, "coinbasevalue", wb->coinbasevalue);

json_set_int(wb_val, "height", wb->height);

json_set_string(wb_val, "flags", wb->flags);

json_set_int(wb_val, "txns", wb->txns);

json_set_string(wb_val, "txn_hashes", wb->txn_hashes);

json_set_int(wb_val, "merkles", wb->merkles);

json_object_set_new_nocheck(wb_val, "merklehash", json_deep_copy(wb->merkle_array));

json_set_string(wb_val, "coinb1", wb->coinb1);

json_set_int(wb_val, "enonce1varlen", wb->enonce1varlen);

json_set_int(wb_val, "enonce2varlen", wb->enonce2varlen);

json_set_int(wb_val, "coinb1len", wb->coinb1len);

json_set_int(wb_val, "coinb2len", wb->coinb2len);

json_set_string(wb_val, "coinb2", wb->coinb2);

ck_rlock(&sdata->instance_lock);

DL_FOREACH2(sdata->node_instances, client, node_next) {

ckmsg_t *client_msg;

smsg_t *msg;

json_t *json_msg = json_deep_copy(wb_val);

json_set_string(json_msg, "node.method", stratum_msgs[SM_WORKINFO]);

client_msg = ckalloc(sizeof(ckmsg_t));

msg = ckzalloc(sizeof(smsg_t));

msg->json_msg = json_msg;

msg->client_id = client->id;

client_msg->data = msg;

DL_APPEND(bulk_send, client_msg);

messages++;

}

DL_FOREACH2(sdata->remote_instances, client, remote_next) {

ckmsg_t *client_msg;

smsg_t *msg;

json_t *json_msg = json_deep_copy(wb_val);

json_set_string(json_msg, "method", stratum_msgs[SM_WORKINFO]);

client_msg = ckalloc(sizeof(ckmsg_t));

msg = ckzalloc(sizeof(smsg_t));

msg->json_msg = json_msg;

msg->client_id = client->id;

client_msg->data = msg;

DL_APPEND(bulk_send, client_msg);

messages++;

}

ck_runlock(&sdata->instance_lock);

if (ckp->remote)

upstream_msgtype(ckp, wb_val, SM_WORKINFO);

json_decref(wb_val);

if (bulk_send) {

LOGINFO("Sending workinfo to mining nodes");

ssend_bulk_append(sdata, bulk_send, messages);

}

}

static json_t *generate_workinfo(ckpool_t *ckp, const workbase_t *wb, const char *func)

{

char cdfield[64];

json_t *val;

sprintf(cdfield, "%lu,%lu", wb->gentime.tv_sec, wb->gentime.tv_nsec);

JSON_CPACK(val, "{sI,ss,ss,ss,ss,ss,ss,ss,ss,sI,so,ss,ss,ss,ss}",

"workinfoid", wb->id,

"poolinstance", ckp->name,

"transactiontree", wb->txn_hashes,

"prevhash", wb->prevhash,

"coinbase1", wb->coinb1,