cpuminer/util.c
pooler 131d5a2374 Add smart handling of long polling failures
Switch to short scan time if long polling fails, and try
to reactivate it only if the server continues to advertise
the feature in HTTP headers.
2012-02-12 03:23:06 +01:00

564 lines
12 KiB
C

/*
* Copyright 2010 Jeff Garzik
*
* 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 2 of the License, or (at your option)
* any later version. See COPYING for more details.
*/
#define _GNU_SOURCE
#include "cpuminer-config.h"
#include <stdio.h>
#include <stdlib.h>
#include <ctype.h>
#include <stdarg.h>
#include <string.h>
#include <unistd.h>
#include <jansson.h>
#include <curl/curl.h>
#include <time.h>
#include "compat.h"
#include "miner.h"
#include "elist.h"
#if JANSSON_MAJOR_VERSION >= 2
#define JSON_LOADS(str, err_ptr) json_loads((str), 0, (err_ptr))
#else
#define JSON_LOADS(str, err_ptr) json_loads((str), (err_ptr))
#endif
struct data_buffer {
void *buf;
size_t len;
};
struct upload_buffer {
const void *buf;
size_t len;
};
struct header_info {
char *lp_path;
};
struct tq_ent {
void *data;
struct list_head q_node;
};
struct thread_q {
struct list_head q;
bool frozen;
pthread_mutex_t mutex;
pthread_cond_t cond;
};
void applog(int prio, const char *fmt, ...)
{
va_list ap;
va_start(ap, fmt);
#ifdef HAVE_SYSLOG_H
if (use_syslog) {
vsyslog(prio, fmt, ap);
}
#else
if (0) {}
#endif
else {
char *f;
int len;
time_t now;
struct tm tm, *tm_p;
time(&now);
pthread_mutex_lock(&time_lock);
tm_p = localtime(&now);
memcpy(&tm, tm_p, sizeof(tm));
pthread_mutex_unlock(&time_lock);
len = 40 + strlen(fmt) + 2;
f = alloca(len);
sprintf(f, "[%d-%02d-%02d %02d:%02d:%02d] %s\n",
tm.tm_year + 1900,
tm.tm_mon + 1,
tm.tm_mday,
tm.tm_hour,
tm.tm_min,
tm.tm_sec,
fmt);
pthread_mutex_lock(&time_lock);
vfprintf(stderr, f, ap); /* atomic write to stderr */
pthread_mutex_unlock(&time_lock);
}
va_end(ap);
}
static void databuf_free(struct data_buffer *db)
{
if (!db)
return;
free(db->buf);
memset(db, 0, sizeof(*db));
}
static size_t all_data_cb(const void *ptr, size_t size, size_t nmemb,
void *user_data)
{
struct data_buffer *db = user_data;
size_t len = size * nmemb;
size_t oldlen, newlen;
void *newmem;
static const unsigned char zero = 0;
oldlen = db->len;
newlen = oldlen + len;
newmem = realloc(db->buf, newlen + 1);
if (!newmem)
return 0;
db->buf = newmem;
db->len = newlen;
memcpy(db->buf + oldlen, ptr, len);
memcpy(db->buf + newlen, &zero, 1); /* null terminate */
return len;
}
static size_t upload_data_cb(void *ptr, size_t size, size_t nmemb,
void *user_data)
{
struct upload_buffer *ub = user_data;
int len = size * nmemb;
if (len > ub->len)
len = ub->len;
if (len) {
memcpy(ptr, ub->buf, len);
ub->buf += len;
ub->len -= len;
}
return len;
}
static size_t resp_hdr_cb(void *ptr, size_t size, size_t nmemb, void *user_data)
{
struct header_info *hi = user_data;
size_t remlen, slen, ptrlen = size * nmemb;
char *rem, *val = NULL, *key = NULL;
void *tmp;
val = calloc(1, ptrlen);
key = calloc(1, ptrlen);
if (!key || !val)
goto out;
tmp = memchr(ptr, ':', ptrlen);
if (!tmp || (tmp == ptr)) /* skip empty keys / blanks */
goto out;
slen = tmp - ptr;
if ((slen + 1) == ptrlen) /* skip key w/ no value */
goto out;
memcpy(key, ptr, slen); /* store & nul term key */
key[slen] = 0;
rem = ptr + slen + 1; /* trim value's leading whitespace */
remlen = ptrlen - slen - 1;
while ((remlen > 0) && (isspace(*rem))) {
remlen--;
rem++;
}
memcpy(val, rem, remlen); /* store value, trim trailing ws */
val[remlen] = 0;
while ((*val) && (isspace(val[strlen(val) - 1]))) {
val[strlen(val) - 1] = 0;
}
if (!*val) /* skip blank value */
goto out;
if (opt_protocol)
applog(LOG_DEBUG, "HTTP hdr(%s): %s", key, val);
if (!strcasecmp("X-Long-Polling", key)) {
hi->lp_path = val; /* steal memory reference */
val = NULL;
}
out:
free(key);
free(val);
return ptrlen;
}
json_t *json_rpc_call(CURL *curl, const char *url,
const char *userpass, const char *rpc_req,
bool longpoll_scan, bool longpoll, int *curl_err)
{
json_t *val, *err_val, *res_val;
int rc;
struct data_buffer all_data = { };
struct upload_buffer upload_data;
json_error_t err = { };
struct curl_slist *headers = NULL;
char len_hdr[64];
char curl_err_str[CURL_ERROR_SIZE];
long timeout = opt_timeout;
struct header_info hi = { };
bool lp_scanning = longpoll_scan && !have_longpoll;
/* it is assumed that 'curl' is freshly [re]initialized at this pt */
if (opt_protocol)
curl_easy_setopt(curl, CURLOPT_VERBOSE, 1);
curl_easy_setopt(curl, CURLOPT_URL, url);
curl_easy_setopt(curl, CURLOPT_ENCODING, "");
curl_easy_setopt(curl, CURLOPT_FAILONERROR, 1);
curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1);
curl_easy_setopt(curl, CURLOPT_TCP_NODELAY, 1);
curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, all_data_cb);
curl_easy_setopt(curl, CURLOPT_WRITEDATA, &all_data);
curl_easy_setopt(curl, CURLOPT_READFUNCTION, upload_data_cb);
curl_easy_setopt(curl, CURLOPT_READDATA, &upload_data);
curl_easy_setopt(curl, CURLOPT_ERRORBUFFER, curl_err_str);
curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1);
curl_easy_setopt(curl, CURLOPT_TIMEOUT, timeout);
if (lp_scanning) {
curl_easy_setopt(curl, CURLOPT_HEADERFUNCTION, resp_hdr_cb);
curl_easy_setopt(curl, CURLOPT_HEADERDATA, &hi);
}
if (userpass) {
curl_easy_setopt(curl, CURLOPT_USERPWD, userpass);
curl_easy_setopt(curl, CURLOPT_HTTPAUTH, CURLAUTH_BASIC);
}
curl_easy_setopt(curl, CURLOPT_POST, 1);
if (opt_protocol)
applog(LOG_DEBUG, "JSON protocol request:\n%s\n", rpc_req);
upload_data.buf = rpc_req;
upload_data.len = strlen(rpc_req);
sprintf(len_hdr, "Content-Length: %lu",
(unsigned long) upload_data.len);
headers = curl_slist_append(headers,
"Content-Type: application/json");
headers = curl_slist_append(headers, len_hdr);
headers = curl_slist_append(headers, "User-Agent: " PACKAGE_STRING);
headers = curl_slist_append(headers,
"X-Mining-Extensions: midstate");
headers = curl_slist_append(headers, "Accept:"); /* disable Accept hdr*/
headers = curl_slist_append(headers, "Expect:"); /* disable Expect hdr*/
curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);
rc = curl_easy_perform(curl);
if (curl_err != NULL)
*curl_err = rc;
if (rc) {
if (!(longpoll && rc == CURLE_OPERATION_TIMEDOUT))
applog(LOG_ERR, "HTTP request failed: %s", curl_err_str);
goto err_out;
}
/* If X-Long-Polling was found, activate long polling */
if (hi.lp_path) {
have_longpoll = true;
tq_push(thr_info[longpoll_thr_id].q, hi.lp_path);
} else
free(hi.lp_path);
hi.lp_path = NULL;
if (!all_data.buf) {
applog(LOG_ERR, "Empty data received in json_rpc_call.");
goto err_out;
}
val = JSON_LOADS(all_data.buf, &err);
if (!val) {
applog(LOG_ERR, "JSON decode failed(%d): %s", err.line, err.text);
goto err_out;
}
if (opt_protocol) {
char *s = json_dumps(val, JSON_INDENT(3));
applog(LOG_DEBUG, "JSON protocol response:\n%s", s);
free(s);
}
/* JSON-RPC valid response returns a non-null 'result',
* and a null 'error'.
*/
res_val = json_object_get(val, "result");
err_val = json_object_get(val, "error");
if (!res_val || json_is_null(res_val) ||
(err_val && !json_is_null(err_val))) {
char *s;
if (err_val)
s = json_dumps(err_val, JSON_INDENT(3));
else
s = strdup("(unknown reason)");
applog(LOG_ERR, "JSON-RPC call failed: %s", s);
free(s);
goto err_out;
}
databuf_free(&all_data);
curl_slist_free_all(headers);
curl_easy_reset(curl);
return val;
err_out:
databuf_free(&all_data);
curl_slist_free_all(headers);
curl_easy_reset(curl);
return NULL;
}
char *bin2hex(const unsigned char *p, size_t len)
{
int i;
char *s = malloc((len * 2) + 1);
if (!s)
return NULL;
for (i = 0; i < len; i++)
sprintf(s + (i * 2), "%02x", (unsigned int) p[i]);
return s;
}
bool hex2bin(unsigned char *p, const char *hexstr, size_t len)
{
while (*hexstr && len) {
char hex_byte[3];
unsigned int v;
if (!hexstr[1]) {
applog(LOG_ERR, "hex2bin str truncated");
return false;
}
hex_byte[0] = hexstr[0];
hex_byte[1] = hexstr[1];
hex_byte[2] = 0;
if (sscanf(hex_byte, "%x", &v) != 1) {
applog(LOG_ERR, "hex2bin sscanf '%s' failed", hex_byte);
return false;
}
*p = (unsigned char) v;
p++;
hexstr += 2;
len--;
}
return (len == 0 && *hexstr == 0) ? true : false;
}
/* Subtract the `struct timeval' values X and Y,
storing the result in RESULT.
Return 1 if the difference is negative, otherwise 0. */
int
timeval_subtract (
struct timeval *result, struct timeval *x, struct timeval *y)
{
/* Perform the carry for the later subtraction by updating Y. */
if (x->tv_usec < y->tv_usec) {
int nsec = (y->tv_usec - x->tv_usec) / 1000000 + 1;
y->tv_usec -= 1000000 * nsec;
y->tv_sec += nsec;
}
if (x->tv_usec - y->tv_usec > 1000000) {
int nsec = (x->tv_usec - y->tv_usec) / 1000000;
y->tv_usec += 1000000 * nsec;
y->tv_sec -= nsec;
}
/* Compute the time remaining to wait.
`tv_usec' is certainly positive. */
result->tv_sec = x->tv_sec - y->tv_sec;
result->tv_usec = x->tv_usec - y->tv_usec;
/* Return 1 if result is negative. */
return x->tv_sec < y->tv_sec;
}
bool fulltest(const unsigned char *hash, const unsigned char *target)
{
unsigned char hash_swap[32], target_swap[32];
uint32_t *hash32 = (uint32_t *) hash_swap;
uint32_t *target32 = (uint32_t *) target_swap;
int i;
bool rc = true;
char *hash_str, *target_str;
swap256(hash_swap, hash);
swap256(target_swap, target);
for (i = 0; i < 32/4; i++) {
uint32_t h32tmp = swab32(hash32[i]);
uint32_t t32tmp = target32[i];
target32[i] = swab32(target32[i]); /* for printing */
if (h32tmp > t32tmp) {
rc = false;
break;
}
if (h32tmp < t32tmp) {
rc = true;
break;
}
}
if (opt_debug) {
hash_str = bin2hex(hash_swap, 32);
target_str = bin2hex(target_swap, 32);
applog(LOG_DEBUG, " Proof: %s\nTarget: %s\nTrgVal? %s",
hash_str,
target_str,
rc ? "YES (hash < target)" :
"no (false positive; hash > target)");
free(hash_str);
free(target_str);
}
return true; /* FIXME: return rc; */
}
struct thread_q *tq_new(void)
{
struct thread_q *tq;
tq = calloc(1, sizeof(*tq));
if (!tq)
return NULL;
INIT_LIST_HEAD(&tq->q);
pthread_mutex_init(&tq->mutex, NULL);
pthread_cond_init(&tq->cond, NULL);
return tq;
}
void tq_free(struct thread_q *tq)
{
struct tq_ent *ent, *iter;
if (!tq)
return;
list_for_each_entry_safe(ent, iter, &tq->q, q_node) {
list_del(&ent->q_node);
free(ent);
}
pthread_cond_destroy(&tq->cond);
pthread_mutex_destroy(&tq->mutex);
memset(tq, 0, sizeof(*tq)); /* poison */
free(tq);
}
static void tq_freezethaw(struct thread_q *tq, bool frozen)
{
pthread_mutex_lock(&tq->mutex);
tq->frozen = frozen;
pthread_cond_signal(&tq->cond);
pthread_mutex_unlock(&tq->mutex);
}
void tq_freeze(struct thread_q *tq)
{
tq_freezethaw(tq, true);
}
void tq_thaw(struct thread_q *tq)
{
tq_freezethaw(tq, false);
}
bool tq_push(struct thread_q *tq, void *data)
{
struct tq_ent *ent;
bool rc = true;
ent = calloc(1, sizeof(*ent));
if (!ent)
return false;
ent->data = data;
INIT_LIST_HEAD(&ent->q_node);
pthread_mutex_lock(&tq->mutex);
if (!tq->frozen) {
list_add_tail(&ent->q_node, &tq->q);
} else {
free(ent);
rc = false;
}
pthread_cond_signal(&tq->cond);
pthread_mutex_unlock(&tq->mutex);
return rc;
}
void *tq_pop(struct thread_q *tq, const struct timespec *abstime)
{
struct tq_ent *ent;
void *rval = NULL;
int rc;
pthread_mutex_lock(&tq->mutex);
if (!list_empty(&tq->q))
goto pop;
if (abstime)
rc = pthread_cond_timedwait(&tq->cond, &tq->mutex, abstime);
else
rc = pthread_cond_wait(&tq->cond, &tq->mutex);
if (rc)
goto out;
if (list_empty(&tq->q))
goto out;
pop:
ent = list_entry(tq->q.next, struct tq_ent, q_node);
rval = ent->data;
list_del(&ent->q_node);
free(ent);
out:
pthread_mutex_unlock(&tq->mutex);
return rval;
}