thodg/cgminer/util.c

• Show log

  Commit

• Hash : bb4bec98
  Author :
  Date : 2012-10-17T09:34:32
  

  Use select on stratum send to make sure the socket is writeable.

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>
  # include <netdb.h>
  #else
  # include <winsock2.h>
  # include <mstcpip.h>
  # include <ws2tcpip.h>
  #endif
  
  #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;
  	char		*stratum_url;
  	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;
  	}
  
  	if (!strcasecmp("X-Stratum", key)) {
  		hi->stratum_url = val;
  		val = NULL;
  	}
  
  out:
  	free(key);
  	free(val);
  	return ptrlen;
  }
  
  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;
  }
  
  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, 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);
  	}
  	if (longpoll)
  		curl_easy_setopt(curl, CURLOPT_SOCKOPTFUNCTION, json_rpc_call_sockopt_cb);
  	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;
  		if (hi.stratum_url) {
  			pool->stratum_url = hi.stratum_url;
  			hi.stratum_url = NULL;
  		}
  	} else {
  		if (hi.lp_path) {
  			free(hi.lp_path);
  			hi.lp_path = NULL;
  		}
  		if (hi.stratum_url) {
  			free(hi.stratum_url);
  			hi.stratum_url = 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;
  }
  
  /* Returns a malloced array string of a binary value of arbitrary length. The
   * array is rounded up to a 4 byte size to appease architectures that need
   * aligned array  sizes */
  char *bin2hex(const unsigned char *p, size_t len)
  {
  	unsigned int i;
  	ssize_t slen;
  	char *s;
  
  	slen = len * 2 + 1;
  	if (slen % 4)
  		slen += 4 - (slen % 4);
  	s = calloc(slen, 1);
  	if (unlikely(!s))
  		quit(1, "Failed to calloc in bin2hex");
  
  	for (i = 0; i < len; i++)
  		sprintf(s + (i * 2), "%02x", (unsigned int) p[i]);
  
  	return s;
  }
  
  /* Does the reverse of bin2hex but does not allocate any ram */
  bool hex2bin(unsigned char *p, const char *hexstr, size_t len)
  {
  	bool ret = false;
  
  	while (*hexstr && len) {
  		char hex_byte[4];
  		unsigned int v;
  
  		if (unlikely(!hexstr[1])) {
  			applog(LOG_ERR, "hex2bin str truncated");
  			return ret;
  		}
  
  		memset(hex_byte, 0, 4);
  		hex_byte[0] = hexstr[0];
  		hex_byte[1] = hexstr[1];
  
  		if (unlikely(sscanf(hex_byte, "%x", &v) != 1)) {
  			applog(LOG_ERR, "hex2bin sscanf '%s' failed", hex_byte);
  			return ret;
  		}
  
  		*p = (unsigned char) v;
  
  		p++;
  		hexstr += 2;
  		len--;
  	}
  
  	if (likely(len == 0 && *hexstr == 0))
  		ret = true;
  	return ret;
  }
  
  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 = NULL;
  	char url_address[256], port[6];
  	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;
  
  	sprintf(url_address, "%.*s", url_len, url_begin);
  
  	if (port_len)
  		snprintf(port, 6, "%.*s", port_len, port_start);
  	else
  		strcpy(port, "80");
  
  	pool->stratum_port = strdup(port);
  
  	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->sockaddr_url = strdup(url_address);
  
  	return true;
  }
  
  /* Send a single command across a socket, appending \n to it */
  static bool __stratum_send(struct pool *pool, char *s, ssize_t len)
  {
  	SOCKETTYPE sock = pool->sock;
  	ssize_t ssent = 0;
  
  	if (opt_protocol)
  		applog(LOG_DEBUG, "SEND: %s", s);
  
  	strcat(s, "\n");
  	len++;
  
  	while (len > 0 ) {
  		struct timeval timeout = {0, 0};
  		size_t sent = 0;
  		fd_set wd;
  
  		FD_ZERO(&wd);
  		FD_SET(sock, &wd);
  		if (select(sock + 1, NULL, &wd, NULL, &timeout) < 1) {
  			applog(LOG_DEBUG, "Write select failed on pool %d sock", pool->pool_no);
  			return false;
  		}
  		if (curl_easy_send(pool->stratum_curl, s + ssent, len, &sent) != CURLE_OK) {
  			applog(LOG_DEBUG, "Failed to curl_easy_send in stratum_send");
  			return false;
  		}
  		ssent += sent;
  		len -= ssent;
  	}
  
  	return true;
  }
  
  bool stratum_send(struct pool *pool, char *s, ssize_t len)
  {
  	bool ret = false;
  
  	mutex_lock(&pool->stratum_lock);
  	if (pool->stratum_active)
  		ret = __stratum_send(pool, s, len);
  	else
  		applog(LOG_DEBUG, "Stratum send failed due to no pool stratum_active");
  	mutex_unlock(&pool->stratum_lock);
  
  	return ret;
  }
  
  #define RECVSIZE 8191
  #define RBUFSIZE (RECVSIZE + 1)
  
  static void clear_sock(struct pool *pool)
  {
  	SOCKETTYPE sock = pool->sock;
  
  	recv(sock, pool->sockbuf, RECVSIZE, MSG_DONTWAIT);
  	strcpy(pool->sockbuf, "");
  }
  
  /* Check to see if Santa's been good to you */
  static bool sock_full(struct pool *pool, bool wait)
  {
  	SOCKETTYPE sock = pool->sock;
  	struct timeval timeout;
  	fd_set rd;
  
  	if (strlen(pool->sockbuf))
  		return true;
  
  	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(struct pool *pool)
  {
  	ssize_t len, buflen;
  	char *tok, *sret = NULL;
  	size_t n;
  
  	if (!strstr(pool->sockbuf, "\n")) {
  		char s[RBUFSIZE];
  		size_t sspace;
  		CURLcode rc;
  
  		if (!sock_full(pool, true)) {
  			applog(LOG_DEBUG, "Timed out waiting for data on sock_full");
  			goto out;
  		}
  		memset(s, 0, RBUFSIZE);
  
  		mutex_lock(&pool->stratum_lock);
  		rc = curl_easy_recv(pool->stratum_curl, s, RECVSIZE, &n);
  		mutex_unlock(&pool->stratum_lock);
  
  		if (rc != CURLE_OK) {
  			applog(LOG_DEBUG, "Failed to recv sock in recv_line");
  			goto out;
  		}
  		/* Prevent buffer overflows, but if 8k is still not enough,
  		 * likely we have had some comms issues and the data is all
  		 * useless anyway */
  		sspace = RECVSIZE - strlen(pool->sockbuf);
  		strncat(pool->sockbuf, s, sspace);
  	}
  
  	buflen = strlen(pool->sockbuf);
  	tok = strtok(pool->sockbuf, "\n");
  	if (!tok) {
  		applog(LOG_DEBUG, "Failed to parse a \\n terminated string in recv_line");
  		goto out;
  	}
  	sret = strdup(tok);
  	len = strlen(sret);
  
  	/* Copy what's left in the buffer after the \n, including the
  	 * terminating \0 */
  	if (buflen > len + 1)
  		memmove(pool->sockbuf, pool->sockbuf + len + 1, buflen - len + 1);
  	else
  		strcpy(pool->sockbuf, "");
  out:
  	if (!sret)
  		clear_sock(pool);
  	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;
  }
  
  static bool parse_reconnect(struct pool *pool, json_t *val)
  {
  	char *url, *port, address[256];
  
  	memset(address, 0, 255);
  	url = (char *)json_string_value(json_array_get(val, 0));
  	if (!url)
  		url = pool->sockaddr_url;
  
  	port = (char *)json_string_value(json_array_get(val, 1));
  	if (!port)
  		port = pool->stratum_port;
  
  	sprintf(address, "%s:%s", url, port);
  
  	if (!extract_sockaddr(pool, address))
  		return false;
  
  	pool->stratum_url = pool->sockaddr_url;
  
  	applog(LOG_NOTICE, "Reconnect requested from pool %d to %s", pool->pool_no, address);
  
  	if (!initiate_stratum(pool) || !auth_stratum(pool))
  		return false;
  
  	return true;
  }
  
  static bool send_version(struct pool *pool, json_t *val)
  {
  	char s[RBUFSIZE];
  	int id = json_integer_value(json_object_get(val, "id"));
  	
  	if (!id)
  		return false;
  
  	sprintf(s, "{\"id\": %d, \"result\": \""PACKAGE"/"VERSION"\", \"error\": null}", id);
  	if (!stratum_send(pool, s, strlen(s)))
  		return false;
  
  	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;
  
  	if (!s)
  		goto out;
  
  	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;
  	}
  
  	if (!strncasecmp(buf, "client.reconnect", 16) && parse_reconnect(pool, params)) {
  		ret = true;
  		goto out;
  	}
  
  	if (!strncasecmp(buf, "client.get_version", 18) && send_version(pool, val)) {
  		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[RBUFSIZE], *sret = NULL;
  	json_error_t err;
  	bool ret = false;
  
  	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, false)) {
  		sret = recv_line(pool);
  		if (!parse_method(pool, sret)) {
  			clear_sock(pool);
  			applog(LOG_INFO, "Failed to parse stratum buffer");
  			free(sret);
  			return ret;
  		}
  		free(sret);
  	}
  
  	if (!stratum_send(pool, s, strlen(s)))
  		goto out;
  
  	sret = recv_line(pool);
  	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);
  
  	return ret;
  }
  
  bool initiate_stratum(struct pool *pool)
  {
  	json_t *val = NULL, *res_val, *err_val;
  	char curl_err_str[CURL_ERROR_SIZE];
  	char s[RBUFSIZE], *sret = NULL;
  	CURL *curl = NULL;
  	json_error_t err;
  	bool ret = false;
  
  	mutex_lock(&pool->stratum_lock);
  	pool->stratum_active = false;
  
  	if (!pool->stratum_curl) {
  		pool->stratum_curl = curl_easy_init();
  		if (unlikely(!pool->stratum_curl))
  			quit(1, "Failed to curl_easy_init in initiate_stratum");
  	}
  	mutex_unlock(&pool->stratum_lock);
  	curl = pool->stratum_curl;
  
  	/* Create a http url for use with curl */
  	memset(s, 0, RBUFSIZE);
  	sprintf(s, "http://%s:%s", pool->sockaddr_url, pool->stratum_port);
  
  	curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, 60);
  	curl_easy_setopt(curl, CURLOPT_ERRORBUFFER, curl_err_str);
  	curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1);
  	curl_easy_setopt(curl, CURLOPT_URL, s);
  	curl_easy_setopt(curl, CURLOPT_TCP_NODELAY, 1);
  	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);
  	}
  	curl_easy_setopt(curl, CURLOPT_CONNECT_ONLY, 1);
  	if (curl_easy_perform(curl)) {
  		applog(LOG_INFO, "Stratum connect failed to pool %d: %s", pool->pool_no, curl_err_str);
  		goto out;
  	}
  	curl_easy_getinfo(curl, CURLINFO_LASTSOCKET, (long *)&pool->sock);
  
  	sprintf(s, "{\"id\": %d, \"method\": \"mining.subscribe\", \"params\": []}", swork_id++);
  
  	if (!__stratum_send(pool, s, strlen(s))) {
  		applog(LOG_DEBUG, "Failed to send s in initiate_stratum");
  		goto out;
  	}
  
  	if (!sock_full(pool, true)) {
  		applog(LOG_DEBUG, "Timed out waiting for response in initiate_stratum");
  		goto out;
  	}
  
  	sret = recv_line(pool);
  	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;
  	}
  
  	pool->nonce1 = json_array_string(res_val, 1);
  	if (!pool->nonce1) {
  		applog(LOG_INFO, "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_INFO, "Failed to get n2size in initiate_stratum");
  		goto out;
  	}
  
  	ret = true;
  out:
  	if (val)
  		json_decref(val);
  
  	if (ret) {
  		if (!pool->stratum_url)
  			pool->stratum_url = pool->sockaddr_url;
  		pool->stratum_active = true;
  		pool->swork.diff = 1;
  		if (opt_protocol) {
  			applog(LOG_DEBUG, "Pool %d confirmed mining.subscribe with extranonce1 %s extran2size %d",
  			       pool->pool_no, pool->nonce1, pool->n2size);
  		}
  	} else {
  		applog(LOG_DEBUG, "Initiate stratum failed, disabling stratum_active");
  		mutex_lock(&pool->stratum_lock);
  		if (curl) {
  			curl_easy_cleanup(curl);
  			pool->stratum_curl = NULL;
  		}
  		mutex_unlock(&pool->stratum_lock);
  	}
  
  	return ret;
  }
  

Download