thodg/cgminer/util.c

• Show log

  Commit

• Hash : 8baac0d6
  Author :
  Date : 2012-09-30T23:03:45
  

  Submit shares from stratum through the abstracted submit share function detecting what message they belong to and showing the data from the associated work, and then deleting it from the hash.

Download

• util.c

• /*
   * Copyright 2011-2012 Con Kolivas
   * 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 3 of the License, or (at your option)
   * any later version.  See COPYING for more details.
   */
  
  #define _GNU_SOURCE
  #include "config.h"
  
  #include <stdio.h>
  #include <stdlib.h>
  #include <ctype.h>
  #include <stdarg.h>
  #include <string.h>
  #include <jansson.h>
  #include <curl/curl.h>
  #include <time.h>
  #include <errno.h>
  #include <unistd.h>
  #include <sys/types.h>
  #ifndef WIN32
  # include <sys/socket.h>
  # include <netinet/in.h>
  # include <netinet/tcp.h>
  #else
  # include <winsock2.h>
  # include <mstcpip.h>
  #endif
  #include <netdb.h>
  
  #include "miner.h"
  #include "elist.h"
  #include "compat.h"
  #include "util.h"
  
  bool successful_connect = false;
  struct timeval nettime;
  
  struct data_buffer {
  	void		*buf;
  	size_t		len;
  };
  
  struct upload_buffer {
  	const void	*buf;
  	size_t		len;
  };
  
  struct header_info {
  	char		*lp_path;
  	int		rolltime;
  	char		*reason;
  	bool		hadrolltime;
  	bool		canroll;
  	bool		hadexpire;
  };
  
  struct tq_ent {
  	void			*data;
  	struct list_head	q_node;
  };
  
  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;
  	unsigned 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-Roll-Ntime", key)) {
  		hi->hadrolltime = true;
  		if (!strncasecmp("N", val, 1))
  			applog(LOG_DEBUG, "X-Roll-Ntime: N found");
  		else {
  			hi->canroll = true;
  
  			/* Check to see if expire= is supported and if not, set
  			 * the rolltime to the default scantime */
  			if (strlen(val) > 7 && !strncasecmp("expire=", val, 7)) {
  				sscanf(val + 7, "%d", &hi->rolltime);
  				hi->hadexpire = true;
  			} else
  				hi->rolltime = opt_scantime;
  			applog(LOG_DEBUG, "X-Roll-Ntime expiry set to %d", hi->rolltime);
  		}
  	}
  
  	if (!strcasecmp("X-Long-Polling", key)) {
  		hi->lp_path = val;	/* steal memory reference */
  		val = NULL;
  	}
  
  	if (!strcasecmp("X-Reject-Reason", key)) {
  		hi->reason = val;	/* steal memory reference */
  		val = NULL;
  	}
  
  out:
  	free(key);
  	free(val);
  	return ptrlen;
  }
  
  #ifdef CURL_HAS_SOCKOPT
  int json_rpc_call_sockopt_cb(void __maybe_unused *userdata, curl_socket_t fd,
  			     curlsocktype __maybe_unused purpose)
  {
  	int tcp_keepidle = 120;
  	int tcp_keepintvl = 120;
  
  #ifndef WIN32
  	int keepalive = 1;
  	int tcp_keepcnt = 5;
  
  	if (unlikely(setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, &keepalive, sizeof(keepalive))))
  		return 1;
  
  # ifdef __linux
  
  	if (unlikely(setsockopt(fd, SOL_TCP, TCP_KEEPCNT, &tcp_keepcnt, sizeof(tcp_keepcnt))))
  		return 1;
  
  	if (unlikely(setsockopt(fd, SOL_TCP, TCP_KEEPIDLE, &tcp_keepidle, sizeof(tcp_keepidle))))
  		return 1;
  
  	if (unlikely(setsockopt(fd, SOL_TCP, TCP_KEEPINTVL, &tcp_keepintvl, sizeof(tcp_keepintvl))))
  		return 1;
  # endif /* __linux */
  # ifdef __APPLE_CC__
  
  	if (unlikely(setsockopt(fd, IPPROTO_TCP, TCP_KEEPALIVE, &tcp_keepintvl, sizeof(tcp_keepintvl))))
  		return 1;
  
  # endif /* __APPLE_CC__ */
  
  #else /* WIN32 */
  
  	struct tcp_keepalive vals;
  	vals.onoff = 1;
  	vals.keepalivetime = tcp_keepidle * 1000;
  	vals.keepaliveinterval = tcp_keepintvl * 1000;
  
  	DWORD outputBytes;
  
  	if (unlikely(WSAIoctl(fd, SIO_KEEPALIVE_VALS, &vals, sizeof(vals), NULL, 0, &outputBytes, NULL, NULL)))
  		return 1;
  
  #endif /* WIN32 */
  
  	return 0;
  }
  #endif
  
  static void last_nettime(struct timeval *last)
  {
  	rd_lock(&netacc_lock);
  	last->tv_sec = nettime.tv_sec;
  	last->tv_usec = nettime.tv_usec;
  	rd_unlock(&netacc_lock);
  }
  
  static void set_nettime(void)
  {
  	wr_lock(&netacc_lock);
  	gettimeofday(&nettime, NULL);
  	wr_unlock(&netacc_lock);
  }
  
  json_t *json_rpc_call(CURL *curl, const char *url,
  		      const char *userpass, const char *rpc_req,
  		      bool probe, bool longpoll, int *rolltime,
  		      struct pool *pool, bool share)
  {
  	long timeout = longpoll ? (60 * 60) : 60;
  	struct data_buffer all_data = {NULL, 0};
  	struct header_info hi = {NULL, 0, NULL, false, false, false};
  	char len_hdr[64], user_agent_hdr[128];
  	char curl_err_str[CURL_ERROR_SIZE];
  	struct curl_slist *headers = NULL;
  	struct upload_buffer upload_data;
  	json_t *val, *err_val, *res_val;
  	bool probing = false;
  	json_error_t err;
  	int rc;
  
  	memset(&err, 0, sizeof(err));
  
  	/* it is assumed that 'curl' is freshly [re]initialized at this pt */
  
  	if (probe)
  		probing = !pool->probed;
  	curl_easy_setopt(curl, CURLOPT_TIMEOUT, timeout);
  
  #if 0 /* Disable curl debugging since it spews to stderr */
  	if (opt_protocol)
  		curl_easy_setopt(curl, CURLOPT_VERBOSE, 1);
  #endif
  	curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1);
  	curl_easy_setopt(curl, CURLOPT_URL, url);
  	curl_easy_setopt(curl, CURLOPT_ENCODING, "");
  	curl_easy_setopt(curl, CURLOPT_FAILONERROR, 1);
  
  	/* Shares are staggered already and delays in submission can be costly
  	 * so do not delay them */
  	if (!opt_delaynet || share)
  		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_HEADERFUNCTION, resp_hdr_cb);
  	curl_easy_setopt(curl, CURLOPT_HEADERDATA, &hi);
  	curl_easy_setopt(curl, CURLOPT_USE_SSL, CURLUSESSL_TRY);
  	if (pool->rpc_proxy) {
  		curl_easy_setopt(curl, CURLOPT_PROXY, pool->rpc_proxy);
  		curl_easy_setopt(curl, CURLOPT_PROXYTYPE, pool->rpc_proxytype);
  	} else if (opt_socks_proxy) {
  		curl_easy_setopt(curl, CURLOPT_PROXY, opt_socks_proxy);
  		curl_easy_setopt(curl, CURLOPT_PROXYTYPE, CURLPROXY_SOCKS4);
  	}
  	if (userpass) {
  		curl_easy_setopt(curl, CURLOPT_USERPWD, userpass);
  		curl_easy_setopt(curl, CURLOPT_HTTPAUTH, CURLAUTH_BASIC);
  	}
  #ifdef CURL_HAS_SOCKOPT
  	if (longpoll)
  		curl_easy_setopt(curl, CURLOPT_SOCKOPTFUNCTION, json_rpc_call_sockopt_cb);
  #endif
  	curl_easy_setopt(curl, CURLOPT_POST, 1);
  
  	if (opt_protocol)
  		applog(LOG_DEBUG, "JSON protocol request:\n%s", rpc_req);
  
  	upload_data.buf = rpc_req;
  	upload_data.len = strlen(rpc_req);
  	sprintf(len_hdr, "Content-Length: %lu",
  		(unsigned long) upload_data.len);
  	sprintf(user_agent_hdr, "User-Agent: %s", PACKAGE_STRING);
  
  	headers = curl_slist_append(headers,
  		"Content-type: application/json");
  	headers = curl_slist_append(headers,
  		"X-Mining-Extensions: longpoll midstate rollntime submitold");
  
  	if (likely(global_hashrate)) {
  		char ghashrate[255];
  
  		sprintf(ghashrate, "X-Mining-Hashrate: %llu", global_hashrate);
  		headers = curl_slist_append(headers, ghashrate);
  	}
  
  	headers = curl_slist_append(headers, len_hdr);
  	headers = curl_slist_append(headers, user_agent_hdr);
  	headers = curl_slist_append(headers, "Expect:"); /* disable Expect hdr*/
  
  	curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);
  
  	if (opt_delaynet) {
  		/* Don't delay share submission, but still track the nettime */
  		if (!share) {
  			long long now_msecs, last_msecs;
  			struct timeval now, last;
  
  			gettimeofday(&now, NULL);
  			last_nettime(&last);
  			now_msecs = (long long)now.tv_sec * 1000;
  			now_msecs += now.tv_usec / 1000;
  			last_msecs = (long long)last.tv_sec * 1000;
  			last_msecs += last.tv_usec / 1000;
  			if (now_msecs > last_msecs && now_msecs - last_msecs < 250) {
  				struct timespec rgtp;
  
  				rgtp.tv_sec = 0;
  				rgtp.tv_nsec = (250 - (now_msecs - last_msecs)) * 1000000;
  				nanosleep(&rgtp, NULL);
  			}
  		}
  		set_nettime();
  	}
  
  	rc = curl_easy_perform(curl);
  	if (rc) {
  		applog(LOG_INFO, "HTTP request failed: %s", curl_err_str);
  		goto err_out;
  	}
  
  	if (!all_data.buf) {
  		applog(LOG_DEBUG, "Empty data received in json_rpc_call.");
  		goto err_out;
  	}
  
  	if (probing) {
  		pool->probed = true;
  		/* If X-Long-Polling was found, activate long polling */
  		if (hi.lp_path) {
  			if (pool->hdr_path != NULL)
  				free(pool->hdr_path);
  			pool->hdr_path = hi.lp_path;
  		} else
  			pool->hdr_path = NULL;
  	} else if (hi.lp_path) {
  		free(hi.lp_path);
  		hi.lp_path = NULL;
  	}
  
  	*rolltime = hi.rolltime;
  	pool->cgminer_pool_stats.rolltime = hi.rolltime;
  	pool->cgminer_pool_stats.hadrolltime = hi.hadrolltime;
  	pool->cgminer_pool_stats.canroll = hi.canroll;
  	pool->cgminer_pool_stats.hadexpire = hi.hadexpire;
  
  	val = JSON_LOADS(all_data.buf, &err);
  	if (!val) {
  		applog(LOG_INFO, "JSON decode failed(%d): %s", err.line, err.text);
  
  		if (opt_protocol)
  			applog(LOG_DEBUG, "JSON protocol response:\n%s", all_data.buf);
  
  		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_INFO, "JSON-RPC call failed: %s", s);
  
  		free(s);
  
  		goto err_out;
  	}
  
  	if (hi.reason) {
  		json_object_set_new(val, "reject-reason", json_string(hi.reason));
  		free(hi.reason);
  		hi.reason = NULL;
  	}
  	successful_connect = true;
  	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);
  	if (!successful_connect)
  		applog(LOG_DEBUG, "Failed to connect in json_rpc_call");
  	curl_easy_setopt(curl, CURLOPT_FRESH_CONNECT, 1);
  	return NULL;
  }
  
  #if (LIBCURL_VERSION_MAJOR == 7 && LIBCURL_VERSION_MINOR >= 10) || (LIBCURL_VERSION_MAJOR > 7)
  static struct {
  	const char *name;
  	curl_proxytype proxytype;
  } proxynames[] = {
  	{ "http:",	CURLPROXY_HTTP },
  #if (LIBCURL_VERSION_MAJOR > 7) || (LIBCURL_VERSION_MINOR > 19) || (LIBCURL_VERSION_MINOR == 19 && LIBCURL_VERSION_PATCH >= 4)
  	{ "http0:",	CURLPROXY_HTTP_1_0 },
  #endif
  #if (LIBCURL_VERSION_MAJOR > 7) || (LIBCURL_VERSION_MINOR > 15) || (LIBCURL_VERSION_MINOR == 15 && LIBCURL_VERSION_PATCH >= 2)
  	{ "socks4:",	CURLPROXY_SOCKS4 },
  #endif
  	{ "socks5:",	CURLPROXY_SOCKS5 },
  #if (LIBCURL_VERSION_MAJOR > 7) || (LIBCURL_VERSION_MINOR >= 18)
  	{ "socks4a:",	CURLPROXY_SOCKS4A },
  	{ "socks5h:",	CURLPROXY_SOCKS5_HOSTNAME },
  #endif
  	{ NULL,	0 }
  };
  #endif
  
  const char *proxytype(curl_proxytype proxytype)
  {
  	int i;
  
  	for (i = 0; proxynames[i].name; i++)
  		if (proxynames[i].proxytype == proxytype)
  			return proxynames[i].name;
  
  	return "invalid";
  }
  
  char *get_proxy(char *url, struct pool *pool)
  {
  	pool->rpc_proxy = NULL;
  
  #if (LIBCURL_VERSION_MAJOR == 7 && LIBCURL_VERSION_MINOR >= 10) || (LIBCURL_VERSION_MAJOR > 7)
  	char *split;
  	int plen, len, i;
  
  	for (i = 0; proxynames[i].name; i++) {
  		plen = strlen(proxynames[i].name);
  		if (strncmp(url, proxynames[i].name, plen) == 0) {
  			if (!(split = strchr(url, '|')))
  				return url;
  
  			*split = '\0';
  			len = split - url;
  			pool->rpc_proxy = malloc(1 + len - plen);
  			if (!(pool->rpc_proxy))
  				quit(1, "Failed to malloc rpc_proxy");
  
  			strcpy(pool->rpc_proxy, url + plen);
  			pool->rpc_proxytype = proxynames[i].proxytype;
  			url = split + 1;
  			break;
  		}
  	}
  #endif
  	return url;
  }
  
  
  char *bin2hex(const unsigned char *p, size_t len)
  {
  	char *s = malloc((len * 2) + 1);
  	unsigned int i;
  
  	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;
  }
  
  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;
  	char *hash_str, *target_str;
  	bool rc = true;
  	int i;
  
  	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 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)
  {
  	mutex_lock(&tq->mutex);
  
  	tq->frozen = frozen;
  
  	pthread_cond_signal(&tq->cond);
  	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);
  
  	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);
  	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;
  
  	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:
  	mutex_unlock(&tq->mutex);
  	return rval;
  }
  
  int thr_info_create(struct thr_info *thr, pthread_attr_t *attr, void *(*start) (void *), void *arg)
  {
  	return pthread_create(&thr->pth, attr, start, arg);
  }
  
  void thr_info_freeze(struct thr_info *thr)
  {
  	struct tq_ent *ent, *iter;
  	struct thread_q *tq;
  
  	if (!thr)
  		return;
  
  	tq = thr->q;
  	if (!tq)
  		return;
  
  	mutex_lock(&tq->mutex);
  	tq->frozen = true;
  	list_for_each_entry_safe(ent, iter, &tq->q, q_node) {
  		list_del(&ent->q_node);
  		free(ent);
  	}
  	mutex_unlock(&tq->mutex);
  }
  
  void thr_info_cancel(struct thr_info *thr)
  {
  	if (!thr)
  		return;
  
  	if (PTH(thr) != 0L) {
  		pthread_cancel(thr->pth);
  		PTH(thr) = 0L;
  	}
  }
  
  /* Provide a ms based sleep that uses nanosleep to avoid poor usleep accuracy
   * on SMP machines */
  void nmsleep(unsigned int msecs)
  {
  	struct timespec twait, tleft;
  	int ret;
  	ldiv_t d;
  
  	d = ldiv(msecs, 1000);
  	tleft.tv_sec = d.quot;
  	tleft.tv_nsec = d.rem * 1000000;
  	do {
  		twait.tv_sec = tleft.tv_sec;
  		twait.tv_nsec = tleft.tv_nsec;
  		ret = nanosleep(&twait, &tleft);
  	} while (ret == -1 && errno == EINTR);
  }
  
  /* Returns the microseconds difference between end and start times as a double */
  double us_tdiff(struct timeval *end, struct timeval *start)
  {
  	return end->tv_sec * 1000000 + end->tv_usec - start->tv_sec * 1000000 - start->tv_usec;
  }
  
  /* Returns the seconds difference between end and start times as a double */
  double tdiff(struct timeval *end, struct timeval *start)
  {
  	return end->tv_sec - start->tv_sec + (end->tv_usec - start->tv_usec) / 1000000.0;
  }
  
  bool extract_sockaddr(struct pool *pool, char *url)
  {
  	char *url_begin, *url_end, *port_start;
  	char *url_address, *port;
  	struct addrinfo hints, *res;
  	int url_len, port_len = 0;
  
  	url_begin = strstr(url, "//");
  	if (!url_begin)
  		url_begin = url;
  	else
  		url_begin += 2;
  	url_end = strstr(url_begin, ":");
  	if (url_end) {
  		url_len = url_end - url_begin;
  		port_len = strlen(url_begin) - url_len - 1;
  		if (port_len < 1)
  			return false;
  		port_start = url_end + 1;
  	} else
  		url_len = strlen(url_begin);
  
  	if (url_len < 1)
  		return false;
  
  	url_address = alloca(url_len + 1);
  	sprintf(url_address, "%.*s", url_len, url_begin);
  
  	if (port_len) {
  		port = alloca(port_len + 1);
  		sprintf(port, "%.*s", port_len, port_start);
  	} else {
  		port = alloca(4);
  		strcpy(port, "80");
  	}
  
  	memset(&hints, 0, sizeof(struct addrinfo));
  	hints.ai_family = AF_UNSPEC;
  	hints.ai_socktype = SOCK_STREAM;
  	hints.ai_protocol = IPPROTO_TCP;
  
  	if (getaddrinfo(url_address, port, &hints, &res)) {
  		applog(LOG_DEBUG, "Failed to extract sock addr");
  		return false;
  	}
  
  	pool->server = (struct sockaddr_in *)res->ai_addr;
  	pool->stratum_url = strdup(url_address);
  	return true;
  }
  
  /* Send a single command across a socket, appending \n to it */
  bool sock_send(int sock, char *s, ssize_t len)
  {
  	ssize_t sent = 0;
  
  	if (opt_protocol)
  		applog(LOG_DEBUG, "SEND: %s", s);
  
  	strcat(s, "\n");
  	len++;
  
  	while (len > 0 ) {
  		sent = send(sock, s + sent, len, 0);
  		if (SOCKETFAIL(sent))
  			return false;
  		len -= sent;
  	}
  	fsync(sock);
  
  	return true;
  }
  
  #define RECVSIZE 8192
  
  static void clear_sock(SOCKETTYPE sock)
  {
  	char *s = alloca(RECVSIZE);
  
  	recv(sock, s, RECVSIZE, MSG_DONTWAIT);
  }
  
  /* Check to see if Santa's been good to you */
  static bool sock_full(SOCKETTYPE sock, bool wait)
  {
  	struct timeval timeout;
  	fd_set rd;
  
  	FD_ZERO(&rd);
  	FD_SET(sock, &rd);
  	timeout.tv_usec = 0;
  	if (wait)
  		timeout.tv_sec = 60;
  	else
  		timeout.tv_sec = 0;
  	if (select(sock + 1, &rd, NULL, NULL, &timeout) > 0)
  		return true;
  	return false;
  }
  
  /* Peeks at a socket to find the first end of line and then reads just that
   * from the socket and returns that as a malloced char */
  char *recv_line(SOCKETTYPE sock)
  {
  	char *sret = NULL, *s, c;
  	ssize_t offset = 0;
  
  	s = alloca(RECVSIZE);
  	if (SOCKETFAIL(recv(sock, s, RECVSIZE, MSG_PEEK))) {
  		applog(LOG_DEBUG, "Failed to recv sock in recv_line");
  		goto out;
  	}
  	sret = strtok(s, "\n");
  	if (!sret) {
  		applog(LOG_DEBUG, "Failed to parse a \\n terminated string in recv_line");
  		goto out;
  	}
  
  	do {
  		read(sock, &c, 1);
  		memcpy(s + offset++, &c, 1);
  	} while (strncmp(&c, "\n", 1));
  	sret = strdup(s);
  	strcpy(sret + offset - 1, "\0");
  out:
  	if (!sret)
  		clear_sock(sock);
  	else if (opt_protocol)
  		applog(LOG_DEBUG, "RECVD: %s", sret);
  	return sret;
  }
  
  /* Extracts a string value from a json array with error checking. To be used
   * when the value of the string returned is only examined and not to be stored.
   * See json_array_string below */
  static char *__json_array_string(json_t *val, unsigned int entry)
  {
  	json_t *arr_entry;
  
  	if (json_is_null(val))
  		return NULL;
  	if (!json_is_array(val))
  		return NULL;
  	if (entry > json_array_size(val))
  		return NULL;
  	arr_entry = json_array_get(val, entry);
  	if (!json_is_string(arr_entry))
  		return NULL;
  
  	return (char *)json_string_value(arr_entry);
  }
  
  /* Creates a freshly malloced dup of __json_array_string */
  static char *json_array_string(json_t *val, unsigned int entry)
  {
  	char *buf = __json_array_string(val, entry);
  
  	if (buf)
  		return strdup(buf);
  	return NULL;
  }
  
  static bool parse_notify(struct pool *pool, json_t *val)
  {
  	char *job_id, *prev_hash, *coinbase1, *coinbase2, *bbversion, *nbit, *ntime;
  	int merkles, i;
  	json_t *arr;
  	bool clean;
  
  	arr = json_array_get(val, 4);
  	if (!arr || !json_is_array(arr))
  		return false;
  
  	merkles = json_array_size(arr);
  
  	job_id = json_array_string(val, 0);
  	prev_hash = json_array_string(val, 1);
  	coinbase1 = json_array_string(val, 2);
  	coinbase2 = json_array_string(val, 3);
  	bbversion = json_array_string(val, 5);
  	nbit = json_array_string(val, 6);
  	ntime = json_array_string(val, 7);
  	clean = json_is_true(json_array_get(val, 8));
  
  	if (!job_id || !prev_hash || !coinbase1 || !coinbase2 || !bbversion || !nbit || !ntime) {
  		/* Annoying but we must not leak memory */
  		if (job_id)
  			free(job_id);
  		if (prev_hash)
  			free(prev_hash);
  		if (coinbase1)
  			free(coinbase1);
  		if (coinbase2)
  			free(coinbase2);
  		if (bbversion)
  			free(bbversion);
  		if (nbit)
  			free(nbit);
  		if (ntime)
  			free(ntime);
  		return false;
  	}
  
  	mutex_lock(&pool->pool_lock);
  	pool->swork.job_id = job_id;
  	pool->swork.prev_hash = prev_hash;
  	pool->swork.coinbase1 = coinbase1;
  	pool->swork.coinbase2 = coinbase2;
  	pool->swork.bbversion = bbversion;
  	pool->swork.nbit = nbit;
  	pool->swork.ntime = ntime;
  	pool->swork.clean = clean;
  	for (i = 0; i < pool->swork.merkles; i++)
  		free(pool->swork.merkle[i]);
  	if (merkles) {
  		pool->swork.merkle = realloc(pool->swork.merkle, sizeof(char *) * merkles + 1);
  		for (i = 0; i < merkles; i++)
  			pool->swork.merkle[i] = json_array_string(arr, i);
  	}
  	pool->swork.merkles = merkles;
  	if (clean)
  		pool->nonce2 = 0;
  	mutex_unlock(&pool->pool_lock);
  
  	if (opt_protocol) {
  		applog(LOG_DEBUG, "job_id: %s", job_id);
  		applog(LOG_DEBUG, "prev_hash: %s", prev_hash);
  		applog(LOG_DEBUG, "coinbase1: %s", coinbase1);
  		applog(LOG_DEBUG, "coinbase2: %s", coinbase2);
  		for (i = 0; i < merkles; i++)
  			applog(LOG_DEBUG, "merkle%d: %s", i, pool->swork.merkle[i]);
  		applog(LOG_DEBUG, "bbversion: %s", bbversion);
  		applog(LOG_DEBUG, "nbit: %s", nbit);
  		applog(LOG_DEBUG, "ntime: %s", ntime);
  		applog(LOG_DEBUG, "clean: %s", clean ? "yes" : "no");
  	}
  
  	/* A notify message is the closest stratum gets to a getwork */
  	pool->getwork_requested++;
  	total_getworks++;
  	return true;
  }
  
  static bool parse_diff(struct pool *pool, json_t *val)
  {
  	int diff;
  
  	diff = json_integer_value(json_array_get(val, 0));
  	if (diff < 1)
  		return false;
  
  	mutex_lock(&pool->pool_lock);
  	pool->swork.diff = diff;
  	mutex_unlock(&pool->pool_lock);
  
  	applog(LOG_DEBUG, "Pool %d difficulty set to %d", pool->pool_no, diff);
  
  	return true;
  }
  
  bool parse_method(struct pool *pool, char *s)
  {
  	json_t *val = NULL, *method, *err_val, *params;
  	json_error_t err;
  	bool ret = false;
  	char *buf;
  
  	val = JSON_LOADS(s, &err);
  	if (!val) {
  		applog(LOG_INFO, "JSON decode failed(%d): %s", err.line, err.text);
  		goto out;
  	}
  
  	method = json_object_get(val, "method");
  	if (!method)
  		goto out;
  	err_val = json_object_get(val, "error");
  	params = json_object_get(val, "params");
  
  	if (err_val && !json_is_null(err_val)) {
  		char *ss;
  
  		if (err_val)
  			ss = json_dumps(err_val, JSON_INDENT(3));
  		else
  			ss = strdup("(unknown reason)");
  
  		applog(LOG_INFO, "JSON-RPC method decode failed: %s", ss);
  
  		free(ss);
  
  		goto out;
  	}
  
  	buf = (char *)json_string_value(method);
  	if (!buf)
  		goto out;
  
  	if (!strncasecmp(buf, "mining.notify", 13) && parse_notify(pool, params)) {
  		ret = true;
  		goto out;
  	}
  
  	if (!strncasecmp(buf, "mining.set_difficulty", 21) && parse_diff(pool, params)) {
  		ret = true;
  		goto out;
  	}
  
  out:
  	if (val)
  		json_decref(val);
  
  	return ret;
  }
  
  bool auth_stratum(struct pool *pool)
  {
  	json_t *val = NULL, *res_val, *err_val;
  	char *s, *sret = NULL;
  	json_error_t err;
  	bool ret = false;
  
  	s = alloca(RECVSIZE);
  	sprintf(s, "{\"id\": %d, \"method\": \"mining.authorize\", \"params\": [\"%s\", \"%s\"]}",
  		swork_id++, pool->rpc_user, pool->rpc_pass);
  
  	/* Parse all data prior sending auth request */
  	while (sock_full(pool->sock, false)) {
  		sret = recv_line(pool->sock);
  		if (!parse_method(pool, sret)) {
  			clear_sock(pool->sock);
  			applog(LOG_WARNING, "Failed to parse stratum buffer");
  			free(sret);
  			return ret;
  		}
  		free(sret);
  	}
  
  	if (!sock_send(pool->sock, s, strlen(s)))
  		goto out;
  
  	sret = recv_line(pool->sock);
  	if (!sret)
  		goto out;
  	val = JSON_LOADS(sret, &err);
  	free(sret);
  	res_val = json_object_get(val, "result");
  	err_val = json_object_get(val, "error");
  
  	if (!res_val || json_is_false(res_val) || (err_val && !json_is_null(err_val)))  {
  		char *ss;
  
  		if (err_val)
  			ss = json_dumps(err_val, JSON_INDENT(3));
  		else
  			ss = strdup("(unknown reason)");
  		applog(LOG_WARNING, "JSON stratum auth failed: %s", ss);
  		free(ss);
  
  		goto out;
  	}
  	ret = true;
  	applog(LOG_INFO, "Stratum authorisation success for pool %d", pool->pool_no);
  out:
  	if (val)
  		json_decref(val);
  
  	pool->stratum_auth = ret;
  
  	return ret;
  }
  
  bool initiate_stratum(struct pool *pool)
  {
  	json_t *val, *res_val, *err_val, *notify_val;
  	char *s, *buf, *sret = NULL;
  	json_error_t err;
  	bool ret = false;
  
  	if (pool->stratum_active)
  		return true;
  
  	s = alloca(RECVSIZE);
  	sprintf(s, "{\"id\": %d, \"method\": \"mining.subscribe\", \"params\": []}", swork_id++);
  
  	pool->sock = socket(AF_INET, SOCK_STREAM, 0);
  	if (pool->sock == INVSOCK)
  		quit(1, "Failed to create pool socket in initiate_stratum");
  	if (SOCKETFAIL(connect(pool->sock, (struct sockaddr *)pool->server, sizeof(struct sockaddr)))) {
  		applog(LOG_DEBUG, "Failed to connect socket to pool");
  		goto out;
  	}
  
  	if (!sock_send(pool->sock, s, strlen(s))) {
  		applog(LOG_DEBUG, "Failed to send s in initiate_stratum");
  		goto out;
  	}
  
  	if (!sock_full(pool->sock, true)) {
  		applog(LOG_DEBUG, "Timed out waiting for response in initiate_stratum");
  		goto out;
  	}
  
  	sret = recv_line(pool->sock);
  	if (!sret)
  		goto out;
  
  	val = JSON_LOADS(sret, &err);
  	free(sret);
  	if (!val) {
  		applog(LOG_INFO, "JSON decode failed(%d): %s", err.line, err.text);
  		goto out;
  	}
  
  	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 *ss;
  
  		if (err_val)
  			ss = json_dumps(err_val, JSON_INDENT(3));
  		else
  			ss = strdup("(unknown reason)");
  
  		applog(LOG_INFO, "JSON-RPC decode failed: %s", ss);
  
  		free(ss);
  
  		goto out;
  	}
  
  	notify_val = json_array_get(res_val, 0);
  	if (!notify_val || json_is_null(notify_val)) {
  		applog(LOG_WARNING, "Failed to parse notify_val in initiate_stratum");
  		goto out;
  	}
  
  	buf = __json_array_string(notify_val, 0);
  	if (!buf || strcasecmp(buf, "mining.notify")) {
  		applog(LOG_WARNING, "Failed to get mining notify in initiate_stratum");
  		goto out;
  	}
  	pool->subscription = json_array_string(notify_val, 1);
  	if (!pool->subscription) {
  		applog(LOG_WARNING, "Failed to get a subscription in initiate_stratum");
  		goto out;
  	}
  
  	pool->nonce1 = json_array_string(res_val, 1);
  	if (!pool->nonce1) {
  		applog(LOG_WARNING, "Failed to get nonce1 in initiate_stratum");
  		goto out;
  	}
  	pool->n2size = json_integer_value(json_array_get(res_val, 2));
  	if (!pool->n2size) {
  		applog(LOG_WARNING, "Failed to get n2size in initiate_stratum");
  		goto out;
  	}
  
  	ret = true;
  out:
  	if (val)
  		json_decref(val);
  
  	if (ret) {
  		pool->stratum_active = true;
  		pool->swork.diff = 1;
  		if (opt_protocol) {
  			applog(LOG_DEBUG, "Pool %d confirmed mining.notify with subscription %s extranonce1 %s extran2size %d",
  			       pool->pool_no, pool->subscription, pool->nonce1, pool->n2size);
  		}
  	} else
  		CLOSESOCKET(pool->sock);
  
  	return ret;
  }
  

Download