Merge branch 'master' of git://github.com/ckolivas/cgminer

api.c

@@ -2454,6 +2454,10 @@ void api(int api_thr_id)
 		CLOSESOCKET(c);
 	}
 die:
+	/* Blank line fix for older compilers since pthread_cleanup_pop is a
+	 * macro that gets confused by a label existing immediately before it
+	 */
+	;
 	pthread_cleanup_pop(true);
 
 	if (opt_debug)

cgminer.c

@@ -378,10 +378,8 @@ static void sharelog(const char*disposition, const struct work*work)
 		applog(LOG_ERR, "sharelog fwrite error");
 }
 
-static void *submit_work_thread(void *userdata);
-static void *get_work_thread(void *userdata);
-
-static void add_pool(void)
+/* Return value is ignored if not called from add_pool_details */
+static struct pool *add_pool(void)
 {
 	struct pool *pool;
 
@@ -392,13 +390,14 @@ static void add_pool(void)
 	pools[total_pools++] = pool;
 	if (unlikely(pthread_mutex_init(&pool->pool_lock, NULL)))
 		quit(1, "Failed to pthread_mutex_init in add_pool");
+	if (unlikely(pthread_cond_init(&pool->cr_cond, NULL)))
+		quit(1, "Failed to pthread_cond_init in add_pool");
+	INIT_LIST_HEAD(&pool->curlring);
+
 	/* Make sure the pool doesn't think we've been idle since time 0 */
 	pool->tv_idle.tv_sec = ~0UL;
 
-	if (unlikely(pthread_create(&pool->submit_thread, NULL, submit_work_thread, (void *)pool)))
-		quit(1, "Failed to create pool submit thread");
-	if (unlikely(pthread_create(&pool->getwork_thread, NULL, get_work_thread, (void *)pool)))
-		quit(1, "Failed to create pool getwork thread");
+	return pool;
 }
 
 /* Pool variant of test and set */
@@ -1957,77 +1956,59 @@ static void sighandler(int __maybe_unused sig)
 	kill_work();
 }
 
-/* One get work thread is created per pool, so as to use one curl handle for
- * all getwork reqeusts from the same pool, minimising connections opened, but
- * separate from the submit work curl handle to not delay share submissions due
- * to getwork traffic */
-static void *get_work_thread(void *userdata)
+/* Called with pool_lock held. Recruit an extra curl if none are available for
+ * this pool. */
+static void recruit_curl(struct pool *pool)
 {
-	struct pool *pool = (struct pool *)userdata;
-	struct workio_cmd *wc;
-	CURL *curl;
+	struct curl_ent *ce = calloc(sizeof(struct curl_ent), 1);
 
-	pthread_detach(pthread_self());
-
-	/* getwork_q memory never freed */
-	pool->getwork_q = tq_new();
-	if (!pool->getwork_q)
-		quit(1, "Failed to tq_new in get_work_thread");
-
-	curl = curl_easy_init();
-	if (unlikely(!curl))
-		quit(1, "Failed to initialise pool getwork CURL");
+	ce->curl = curl_easy_init();
+	if (unlikely(!ce->curl || !ce))
+		quit(1, "Failed to init in recruit_curl");
 
-	while ((wc = tq_pop(pool->getwork_q, NULL)) != NULL) {
-		struct work *ret_work;
-		int failures = 0;
+	list_add(&ce->node, &pool->curlring);
+	pool->curls++;
+	applog(LOG_DEBUG, "Recruited curl %d for pool %d", pool->curls, pool->pool_no);
+}
 
-		ret_work = make_work();
+/* Grab an available curl if there is one. If not, then recruit extra curls
+ * unless we are in a submit_fail situation, or we have opt_delaynet enabled
+ * and there are already 5 curls in circulation */
+static struct curl_ent *pop_curl_entry(struct pool *pool)
+{
+	struct curl_ent *ce;
 
-		if (wc->thr)
-			ret_work->thr = wc->thr;
+	mutex_lock(&pool->pool_lock);
+	if (!pool->curls)
+		recruit_curl(pool);
+	else if (list_empty(&pool->curlring)) {
+		if ((pool->submit_fail || opt_delaynet) && pool->curls > 4)
+			pthread_cond_wait(&pool->cr_cond, &pool->pool_lock);
 		else
-			ret_work->thr = NULL;
-
-		ret_work->pool = pool;
-
-		/* obtain new work from bitcoin via JSON-RPC */
-		while (!get_upstream_work(ret_work, curl)) {
-			if (unlikely((opt_retries >= 0) && (++failures > opt_retries))) {
-				applog(LOG_ERR, "json_rpc_call failed, terminating workio thread");
-				free_work(ret_work);
-				kill_work();
-				break;
-			}
-
-			/* pause, then restart work-request loop */
-			applog(LOG_DEBUG, "json_rpc_call failed on get work, retry after %d seconds",
-				fail_pause);
-			sleep(fail_pause);
-			fail_pause += opt_fail_pause;
-		}
-		fail_pause = opt_fail_pause;
-
-		applog(LOG_DEBUG, "Pushing work to requesting thread");
-
-		/* send work to requesting thread */
-		if (unlikely(!tq_push(thr_info[stage_thr_id].q, ret_work))) {
-			applog(LOG_ERR, "Failed to tq_push work in workio_get_work");
-			kill_work();
-			free_work(ret_work);
-		}
-		workio_cmd_free(wc);
+			recruit_curl(pool);
 	}
+	ce = list_entry(pool->curlring.next, struct curl_ent, node);
+	list_del(&ce->node);
+	mutex_unlock(&pool->pool_lock);
 
-	curl_easy_cleanup(curl);
-	return NULL;
+	return ce;
 }
 
-static void *get_extra_work(void *userdata)
+static void push_curl_entry(struct curl_ent *ce, struct pool *pool)
+{
+	mutex_lock(&pool->pool_lock);
+	list_add_tail(&ce->node, &pool->curlring);
+	gettimeofday(&ce->tv, NULL);
+	pthread_cond_signal(&pool->cr_cond);
+	mutex_unlock(&pool->pool_lock);
+}
+
+static void *get_work_thread(void *userdata)
 {
 	struct workio_cmd *wc = (struct workio_cmd *)userdata;
-	struct work *ret_work = make_work();;
-	CURL *curl = curl_easy_init();
+	struct work *ret_work = make_work();
+	struct curl_ent *ce;
+	struct pool *pool;
 	int failures = 0;
 
 	pthread_detach(pthread_self());
@@ -2039,10 +2020,11 @@ static void *get_extra_work(void *userdata)
 	else
 		ret_work->thr = NULL;
 
-	ret_work->pool = select_pool(wc->lagging);
+	pool = ret_work->pool = select_pool(wc->lagging);
+	ce = pop_curl_entry(pool);
 
 	/* obtain new work from bitcoin via JSON-RPC */
-	while (!get_upstream_work(ret_work, curl)) {
+	while (!get_upstream_work(ret_work, ce->curl)) {
 		if (unlikely((opt_retries >= 0) && (++failures > opt_retries))) {
 			applog(LOG_ERR, "json_rpc_call failed, terminating workio thread");
 			free_work(ret_work);
@@ -2069,7 +2051,7 @@ static void *get_extra_work(void *userdata)
 
 out:
 	workio_cmd_free(wc);
-	curl_easy_cleanup(curl);
+	push_curl_entry(ce, pool);
 	return NULL;
 }
 
@@ -2078,13 +2060,9 @@ out:
  * requests */
 static bool workio_get_work(struct workio_cmd *wc)
 {
-	struct pool *pool = select_pool(wc->lagging);
 	pthread_t get_thread;
 
-	if (list_empty(&pool->getwork_q->q) || pool->submit_fail)
-		return tq_push(pool->getwork_q, wc);
-
-	if (unlikely(pthread_create(&get_thread, NULL, get_extra_work, (void *)wc))) {
+	if (unlikely(pthread_create(&get_thread, NULL, get_work_thread, (void *)wc))) {
 		applog(LOG_ERR, "Failed to create get_work_thread");
 		return false;
 	}
@@ -2114,83 +2092,13 @@ static bool stale_work(struct work *work, bool share)
 	return false;
 }
 
-/* One submit work thread is created per pool, so as to use one curl handle
- * for all submissions to the same pool, minimising connections opened, but
- * separate from the getwork curl handle to not delay share submission due to
- * getwork traffic */
-static void *submit_work_thread(void *userdata)
-{
-	struct pool *pool = (struct pool *)userdata;
-	struct workio_cmd *wc;
-	CURL *curl;
-
-	pthread_detach(pthread_self());
 
-	/* submit_q memory never freed */
-	pool->submit_q = tq_new();
-	if (!pool->submit_q )
-		quit(1, "Failed to tq_new in submit_work_thread");
-
-	curl = curl_easy_init();
-	if (unlikely(!curl))
-		quit(1, "Failed to initialise pool submit CURL");
-
-	while ((wc = tq_pop(pool->submit_q, NULL)) != NULL) {
-		struct work *work = wc->u.work;
-		int failures = 0;
-
-		if (stale_work(work, true)) {
-			if (pool->submit_old)
-				applog(LOG_NOTICE, "Stale share, submitting as pool %d requested",
-				       pool->pool_no);
-			else if (opt_submit_stale)
-				applog(LOG_NOTICE, "Stale share from pool %d, submitting as user requested",
-					pool->pool_no);
-			else {
-				applog(LOG_NOTICE, "Stale share from pool %d, discarding",
-					pool->pool_no);
-				sharelog("discard", work);
-				total_stale++;
-				pool->stale_shares++;
-				workio_cmd_free(wc);
-				continue;
-			}
-		}
-
-		/* submit solution to bitcoin via JSON-RPC */
-		while (!submit_upstream_work(work, curl)) {
-			if (stale_work(work, true)) {
-				applog(LOG_NOTICE, "Share became stale while retrying submit, discarding");
-				total_stale++;
-				pool->stale_shares++;
-				break;
-			}
-			if (unlikely((opt_retries >= 0) && (++failures > opt_retries))) {
-				applog(LOG_ERR, "Failed %d retries ...terminating workio thread", opt_retries);
-				kill_work();
-				break;
-			}
-
-			/* pause, then restart work-request loop */
-			applog(LOG_INFO, "json_rpc_call failed on submit_work, retry after %d seconds",
-				fail_pause);
-			sleep(fail_pause);
-			fail_pause += opt_fail_pause;
-		}
-		fail_pause = opt_fail_pause;
-		workio_cmd_free(wc);
-	}
-
-	curl_easy_cleanup(curl);
-	return NULL;
-}
-
-static void *submit_extra_work(void *userdata)
+static void *submit_work_thread(void *userdata)
 {
 	struct workio_cmd *wc = (struct workio_cmd *)userdata;
 	struct work *work = wc->u.work;
 	struct pool *pool = work->pool;
-	CURL *curl = curl_easy_init();
+	struct curl_ent *ce;
 	int failures = 0;
 
 	pthread_detach(pthread_self());
@@ -2211,8 +2119,9 @@ static void *submit_extra_work(void *userdata)
 		}
 	}
 
+	ce = pop_curl_entry(pool);
 	/* submit solution to bitcoin via JSON-RPC */
-	while (!submit_upstream_work(work, curl)) {
+	while (!submit_upstream_work(work, ce->curl)) {
 		if (stale_work(work, true)) {
 			applog(LOG_NOTICE, "Share became stale while retrying submit, discarding");
 			total_stale++;
@@ -2232,9 +2141,9 @@ static void *submit_extra_work(void *userdata)
 		fail_pause += opt_fail_pause;
 	}
 	fail_pause = opt_fail_pause;
+	push_curl_entry(ce, pool);
 out:
 	workio_cmd_free(wc);
-	curl_easy_cleanup(curl);
 	return NULL;
 }
 
@@ -2244,13 +2153,9 @@ out:
  * any size hardware */
 static bool workio_submit_work(struct workio_cmd *wc)
 {
-	struct pool *pool = wc->u.work->pool;
 	pthread_t submit_thread;
 
-	if (list_empty(&pool->submit_q->q) || pool->submit_fail)
-		return tq_push(pool->submit_q, wc);
-
-	if (unlikely(pthread_create(&submit_thread, NULL, submit_extra_work, (void *)wc))) {
+	if (unlikely(pthread_create(&submit_thread, NULL, submit_work_thread, (void *)wc))) {
 		applog(LOG_ERR, "Failed to create submit_work_thread");
 		return false;
 	}
@@ -4097,6 +4002,26 @@ void reinit_device(struct cgpu_info *cgpu)
 
 static struct timeval rotate_tv;
 
+/* We reap curls if they are unused for over a minute */
+static void reap_curl(struct pool *pool)
+{
+	struct curl_ent *ent, *iter;
+	struct timeval now;
+
+	gettimeofday(&now, NULL);
+	mutex_lock(&pool->pool_lock);
+	list_for_each_entry_safe(ent, iter, &pool->curlring, node) {
+		if (now.tv_sec - ent->tv.tv_sec > 60) {
+			applog(LOG_DEBUG, "Reaped curl %d from pool %d", pool->curls, pool->pool_no);
+			pool->curls--;
+			list_del(&ent->node);
+			curl_easy_cleanup(ent->curl);
+			free(ent);
+		}
+	}
+	mutex_unlock(&pool->pool_lock);
+}
+
 static void *watchpool_thread(void __maybe_unused *userdata)
 {
 	pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);
@@ -4110,6 +4035,7 @@ static void *watchpool_thread(void __maybe_unused *userdata)
 		for (i = 0; i < total_pools; i++) {
 			struct pool *pool = pools[i];
 
+			reap_curl(pool);
 			if (!pool->enabled)
 				continue;
 
@@ -4126,7 +4052,7 @@ static void *watchpool_thread(void __maybe_unused *userdata)
 			switch_pools(NULL);
 		}
 
-		sleep(10);
+		sleep(30);
 	}
 	return NULL;
 }
@@ -4442,18 +4368,13 @@ char *curses_input(const char *query)
 
 int add_pool_details(bool live, char *url, char *user, char *pass)
 {
-	struct pool *pool = NULL;
+	struct pool *pool;
 
 	if (total_pools == MAX_POOLS)
 		return ADD_POOL_MAXIMUM;
 
-	pool = calloc(sizeof(struct pool), 1);
-	if (!pool)
-		quit(1, "Failed to realloc pools in add_pool_details");
-	pool->pool_no = total_pools;
-	pool->prio = total_pools;
-	if (unlikely(pthread_mutex_init(&pool->pool_lock, NULL)))
-		quit (1, "Failed to pthread_mutex_init in input_pool");
+	pool = add_pool();
+
 	pool->rpc_url = url;
 	pool->rpc_user = user;
 	pool->rpc_pass = pass;
@@ -4462,19 +4383,11 @@ int add_pool_details(bool live, char *url, char *user, char *pass)
 		quit(1, "Failed to malloc userpass");
 	sprintf(pool->rpc_userpass, "%s:%s", pool->rpc_user, pool->rpc_pass);
 
-	pool->tv_idle.tv_sec = ~0UL;
-
-	if (unlikely(pthread_create(&pool->submit_thread, NULL, submit_work_thread, (void *)pool)))
-		quit(1, "Failed to create pool submit thread");
-	if (unlikely(pthread_create(&pool->getwork_thread, NULL, get_work_thread, (void *)pool)))
-		quit(1, "Failed to create pool getwork thread");
-
 	/* Test the pool is not idle if we're live running, otherwise
 	 * it will be tested separately */
 	pool->enabled = true;
 	if (live && !pool_active(pool, false))
 		pool->idle = true;
-	pools[total_pools++] = pool;
 
 	return ADD_POOL_OK;
 }

miner.h

@@ -598,6 +598,12 @@ typedef struct {
 } dev_blk_ctx;
 #endif
 
+struct curl_ent {
+	CURL *curl;
+	struct list_head node;
+	struct timeval tv;
+};
+
 struct pool {
 	int pool_no;
 	int prio;
@@ -634,6 +640,10 @@ struct pool {
 	pthread_t longpoll_thread;
 	pthread_t submit_thread;
 	pthread_t getwork_thread;
+
+	int curls;
+	pthread_cond_t cr_cond;
+	struct list_head curlring;
 };
 
 struct work {