Merge branch 'master' of https://bitbucket.org/ckolivas/ckpool

configure.ac

@@ -1,4 +1,4 @@
-AC_INIT(ckpool, 0.8, kernel@kolivas.org)
+AC_INIT(ckpool, 0.8.1, kernel@kolivas.org)
 
 AC_CANONICAL_SYSTEM
 AC_CONFIG_MACRO_DIR([m4])

src/connector.c

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 Con Kolivas
+ * Copyright 2014-2015 Con Kolivas
  *
  * This program is free software; you can redistribute it and/or modify it
  * under the terms of the GNU General Public License as published by the Free
@@ -236,13 +236,14 @@ static void stratifier_drop_client(ckpool_t *ckp, int64_t id)
  * regularly but keep the instances in a linked list until their ref count
  * drops to zero when we can remove them lazily. Client must hold a reference
  * count. */
-static void invalidate_client(ckpool_t *ckp, cdata_t *cdata, client_instance_t *client)
+static int invalidate_client(ckpool_t *ckp, cdata_t *cdata, client_instance_t *client)
 {
 	client_instance_t *tmp;
+	int ret;
 
-	drop_client(cdata, client);
+	ret = drop_client(cdata, client);
 	if (ckp->passthrough)
-		return;
+		goto out;
 	stratifier_drop_client(ckp, client->id);
 
 	/* Cull old unused clients lazily when there are no more reference
@@ -256,6 +257,9 @@ static void invalidate_client(ckpool_t *ckp, cdata_t *cdata, client_instance_t *
 		}
 	}
 	ck_wunlock(&cdata->lock);
+
+out:
+	return ret;
 }
 
 static void send_client(cdata_t *cdata, int64_t id, char *buf);
@@ -618,8 +622,7 @@ static void process_client_msg(cdata_t *cdata, const char *buf)
 		json_object_set_new_nocheck(json_msg, "client_id", json_integer(passthrough_id));
 	} else
 		client_id = client_id64;
-	msg = json_dumps(json_msg, 0);
+	msg = json_dumps(json_msg, JSON_EOL);
-	realloc_strcat(&msg, "\n");
 	send_client(cdata, client_id, msg);
 	json_decref(json_msg);
 }
@@ -675,19 +678,6 @@ retry:
 	 * so look for them first. */
 	if (likely(buf[0] == '{')) {
 		process_client_msg(cdata, buf);
-	} else if (cmdmatch(buf, "ping")) {
-		LOGDEBUG("Connector received ping request");
-		send_unix_msg(sockd, "pong");
-	} else if (cmdmatch(buf, "accept")) {
-		LOGDEBUG("Connector received accept signal");
-		cdata->accept = true;
-	} else if (cmdmatch(buf, "reject")) {
-		LOGDEBUG("Connector received reject signal");
-		cdata->accept = false;
-	} else if (cmdmatch(buf, "loglevel")) {
-		sscanf(buf, "loglevel=%d", &ckp->loglevel);
-	} else if (cmdmatch(buf, "shutdown")) {
-		goto out;
 	} else if (cmdmatch(buf, "dropclient")) {
 		client_instance_t *client;
 
@@ -700,12 +690,26 @@ retry:
 		client = ref_client_by_id(cdata, client_id);
 		if (unlikely(!client)) {
 			LOGINFO("Connector failed to find client id %ld to drop", client_id);
+			stratifier_drop_client(ckp, client_id);
 			goto retry;
 		}
-		ret = drop_client(cdata, client);
+		ret = invalidate_client(ckp, cdata, client);
 		dec_instance_ref(cdata, client);
 		if (ret >= 0)
 			LOGINFO("Connector dropped client id: %ld", client_id);
+	} else if (cmdmatch(buf, "ping")) {
+		LOGDEBUG("Connector received ping request");
+		send_unix_msg(sockd, "pong");
+	} else if (cmdmatch(buf, "accept")) {
+		LOGDEBUG("Connector received accept signal");
+		cdata->accept = true;
+	} else if (cmdmatch(buf, "reject")) {
+		LOGDEBUG("Connector received reject signal");
+		cdata->accept = false;
+	} else if (cmdmatch(buf, "loglevel")) {
+		sscanf(buf, "loglevel=%d", &ckp->loglevel);
+	} else if (cmdmatch(buf, "shutdown")) {
+		goto out;
 	} else if (cmdmatch(buf, "passthrough")) {
 		client_instance_t *client;
 

src/generator.c

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 Con Kolivas
+ * Copyright 2014-2015 Con Kolivas
  *
  * This program is free software; you can redistribute it and/or modify it
  * under the terms of the GNU General Public License as published by the Free
@@ -388,13 +388,11 @@ static bool send_json_msg(connsock_t *cs, json_t *json_msg)
 	int len, sent;
 	char *s;
 
-	s = json_dumps(json_msg, JSON_ESCAPE_SLASH);
+	s = json_dumps(json_msg, JSON_ESCAPE_SLASH | JSON_EOL);
 	LOGDEBUG("Sending json msg: %s", s);
 	len = strlen(s);
 	sent = write_socket(cs->fd, s, len);
-	len += 1;
 	dealloc(s);
-	sent += write_socket(cs->fd, "\n", 1);
 	if (sent != len) {
 		LOGWARNING("Failed to send %d bytes sent %d in send_json_msg", len, sent);
 		return false;

src/jansson-2.6/src/dump.c

@@ -427,6 +427,13 @@ static int do_dump(const json_t *json, size_t flags, int depth,
     }
 }
 
+char *json_dump_dup(const char *str, size_t flags)
+{
+	if (flags & JSON_EOL)
+		return jsonp_eolstrdup(str);
+	return jsonp_strdup(str);
+}
+
 char *json_dumps(const json_t *json, size_t flags)
 {
     strbuffer_t strbuff;
@@ -438,7 +445,7 @@ char *json_dumps(const json_t *json, size_t flags)
     if(json_dump_callback(json, dump_to_strbuffer, (void *)&strbuff, flags))
         result = NULL;
     else
-        result = jsonp_strdup(strbuffer_value(&strbuff));
+        result = json_dump_dup(strbuffer_value(&strbuff), flags);
 
     strbuffer_close(&strbuff);
     return result;

src/jansson-2.6/src/jansson.h

@@ -261,6 +261,7 @@ json_t *json_load_callback(json_load_callback_t callback, void *data, size_t fla
 #define JSON_ENCODE_ANY     0x200
 #define JSON_ESCAPE_SLASH   0x400
 #define JSON_NO_UTF8        0x800
+#define JSON_EOL            0x1000
 
 typedef int (*json_dump_callback_t)(const char *buffer, size_t size, void *data);
 

src/jansson-2.6/src/jansson_private.h

@@ -1,5 +1,6 @@
 /*
  * Copyright (c) 2009-2013 Petri Lehtinen <petri@digip.org>
+ * Copyright (c) 2015 Con Kolivas <kernel@kolivas.org>
  *
  * Jansson is free software; you can redistribute it and/or modify
  * it under the terms of the MIT license. See LICENSE for details.
@@ -83,6 +84,7 @@ void* jsonp_malloc(size_t size);
 void jsonp_free(void *ptr);
 char *jsonp_strndup(const char *str, size_t length);
 char *jsonp_strdup(const char *str);
+char *jsonp_eolstrdup(const char *str);
 
 /* Windows compatibility */
 #ifdef _WIN32

src/jansson-2.6/src/memory.c

@@ -1,6 +1,7 @@
 /*
  * Copyright (c) 2009-2013 Petri Lehtinen <petri@digip.org>
  * Copyright (c) 2011-2012 Basile Starynkevitch <basile@starynkevitch.net>
+ * Copyright (c) 2015 Con Kolivas <kernel@kolivas.org>
  *
  * Jansson is free software; you can redistribute it and/or modify it
  * under the terms of the MIT license. See LICENSE for details.
@@ -49,6 +50,25 @@ char *jsonp_strdup(const char *str)
     return new_str;
 }
 
+char *jsonp_eolstrdup(const char *str)
+{
+    char *new_str;
+    size_t len;
+
+    len = strlen(str);
+    if(len == (size_t)-1)
+        return NULL;
+
+    new_str = jsonp_malloc(len + 2);
+    if(!new_str)
+        return NULL;
+
+    memcpy(new_str, str, len);
+    new_str[len] = '\n';
+    new_str[len + 1] = '\0';
+    return new_str;
+}
+
 void json_set_alloc_funcs(json_malloc_t malloc_fn, json_free_t free_fn)
 {
     do_malloc = malloc_fn;

src/stratifier.c

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 Con Kolivas
+ * Copyright 2014-2015 Con Kolivas
  *
  * This program is free software; you can redistribute it and/or modify it
  * under the terms of the GNU General Public License as published by the Free
@@ -172,6 +172,8 @@ struct user_instance {
 
 	int workers;
 
+	double best_diff; /* Best share found by this user */
+
 	double dsps1; /* Diff shares per second, 1 minute rolling average */
 	double dsps5; /* ... 5 minute ... */
 	double dsps60;/* etc */
@@ -195,6 +197,7 @@ struct worker_instance {
 	tv_t last_share;
 	time_t start_time;
 
+	double best_diff; /* Best share found by this worker */
 	int mindiff; /* User chosen mindiff */
 
 	bool idle;
@@ -252,8 +255,10 @@ struct stratum_instance {
 	ckpool_t *ckp;
 
 	time_t last_txns; /* Last time this worker requested txn hashes */
+	time_t disconnected_time; /* Time this instance disconnected */
 
 	int64_t suggest_diff; /* Stratum client suggested diff */
+	double best_diff; /* Best share found by this instance */
 };
 
 struct share {
@@ -1098,6 +1103,14 @@ static stratum_instance_t *__stratum_add_instance(ckpool_t *ckp, int64_t id, int
 	return instance;
 }
 
+/* Add a stratum instance to the dead instances list */
+static void kill_instance(sdata_t *sdata, stratum_instance_t *client)
+{
+	if (client->user_instance)
+		DL_DELETE(client->user_instance->instances, client);
+	LL_PREPEND(sdata->dead_instances, client);
+}
+
 /* Only supports a full ckpool instance sessionid with an 8 byte sessionid */
 static bool disconnected_sessionid_exists(sdata_t *sdata, const char *sessionid, int64_t id)
 {
@@ -1112,7 +1125,7 @@ static bool disconnected_sessionid_exists(sdata_t *sdata, const char *sessionid,
 	/* Number is in BE but we don't swap either of them */
 	hex2bin(&session64, sessionid, 8);
 
-	ck_rlock(&sdata->instance_lock);
+	ck_ilock(&sdata->instance_lock);
 	HASH_ITER(hh, sdata->stratum_instances, instance, tmp) {
 		if (instance->id == id)
 			continue;
@@ -1123,10 +1136,18 @@ static bool disconnected_sessionid_exists(sdata_t *sdata, const char *sessionid,
 	}
 	instance = NULL;
 	HASH_FIND(hh, sdata->disconnected_instances, &session64, sizeof(uint64_t), instance);
-	if (instance)
+	if (instance) {
+		/* If we've found a matching disconnected instance, use it only
+		 * once and discard it */
+		ck_ulock(&sdata->instance_lock);
+		HASH_DEL(sdata->disconnected_instances, instance);
+		kill_instance(sdata, instance);
+		ck_dwilock(&sdata->instance_lock);
+
 		ret = true;
+	}
 out_unlock:
-	ck_runlock(&sdata->instance_lock);
+	ck_uilock(&sdata->instance_lock);
 out:
 	return ret;
 }
@@ -1209,6 +1230,7 @@ static void dec_worker(ckpool_t *ckp, user_instance_t *instance)
 static void drop_client(sdata_t *sdata, int64_t id)
 {
 	stratum_instance_t *client, *tmp;
+	time_t now_t = time(NULL);
 	bool dec = false;
 
 	LOGINFO("Stratifier dropping client %ld", id);
@@ -1227,13 +1249,11 @@ static void drop_client(sdata_t *sdata, int64_t id)
 		HASH_DEL(sdata->stratum_instances, client);
 		HASH_FIND(hh, sdata->disconnected_instances, &client->enonce1_64, sizeof(uint64_t), old_client);
 		/* Only keep around one copy of the old client in server mode */
-		if (!client->ckp->proxy && !old_client && client->enonce1_64)
+		if (!client->ckp->proxy && !old_client && client->enonce1_64) {
 			HASH_ADD(hh, sdata->disconnected_instances, enonce1_64, sizeof(uint64_t), client);
-		else {
+			client->disconnected_time = time(NULL);
-			if (client->user_instance)
+		} else
-				DL_DELETE(client->user_instance->instances, client);
+			kill_instance(sdata, client);
-			LL_PREPEND(sdata->dead_instances, client);
-		}
 	}
 	ck_wunlock(&sdata->instance_lock);
 
@@ -1247,6 +1267,18 @@ static void drop_client(sdata_t *sdata, int64_t id)
 	ck_wlock(&sdata->instance_lock);
 	if (client)
 		__dec_instance_ref(client);
+	/* Old disconnected instances will not have any valid shares so remove
+	 * them from the disconnected instances list if they've been dead for
+	 * more than 10 minutes */
+	HASH_ITER(hh, sdata->disconnected_instances, client, tmp) {
+		if (now_t - client->disconnected_time < 600)
+			continue;
+		LOGINFO("Discarding aged disconnected instance %ld", client->id);
+		HASH_DEL(sdata->disconnected_instances, client);
+		kill_instance(sdata, client);
+	}
+	/* Discard any dead instances that no longer hold any reference counts,
+	 * freeing up their memory safely */
 	DL_FOREACH_SAFE(sdata->dead_instances, client, tmp) {
 		if (!client->ref) {
 			LOGINFO("Stratifier discarding instance %ld", client->id);
@@ -2533,6 +2565,17 @@ static json_t *parse_submit(stratum_instance_t *client, json_t *json_msg,
 		nonce2[len] = '\0';
 	}
 	sdiff = submission_diff(client, wb, nonce2, ntime32, nonce, hash);
+	if (sdiff > client->best_diff) {
+		worker_instance_t *worker = client->worker_instance;
+
+		client->best_diff = sdiff;
+		LOGINFO("User %s worker %s client %ld new best diff %lf", user_instance->username,
+			worker->workername, client->id, sdiff);
+		if (sdiff > worker->best_diff)
+			worker->best_diff = sdiff;
+		if (sdiff > user_instance->best_diff)
+			user_instance->best_diff = sdiff;
+	}
 	bswap_256(sharehash, hash);
 	__bin2hex(hexhash, sharehash, 32);
 
@@ -2619,9 +2662,9 @@ out_unlock:
 	if (ckp->logshares) {
 		fp = fopen(fname, "ae");
 		if (likely(fp)) {
-			s = json_dumps(val, 0);
+			s = json_dumps(val, JSON_EOL);
 			len = strlen(s);
-			len = fprintf(fp, "%s\n", s);
+			len = fprintf(fp, "%s", s);
 			free(s);
 			fclose(fp);
 			if (unlikely(len < 0))
@@ -3543,11 +3586,12 @@ static void *statsupdate(void *arg)
 				ghs = worker->dsps1440 * nonces;
 				suffix_string(ghs, suffix1440, 16, 0);
 
-				JSON_CPACK(val, "{ss,ss,ss,ss}",
+				JSON_CPACK(val, "{ss,ss,ss,ss,sf}",
 						"hashrate1m", suffix1,
 						"hashrate5m", suffix5,
 						"hashrate1hr", suffix60,
-						"hashrate1d", suffix1440);
+						"hashrate1d", suffix1440,
+						"bestshare", worker->best_diff);
 
 				snprintf(fname, 511, "%s/workers/%s", ckp->logdir, worker->workername);
 				fp = fopen(fname, "we");
@@ -3555,8 +3599,8 @@ static void *statsupdate(void *arg)
 					LOGERR("Failed to fopen %s", fname);
 					continue;
 				}
-				s = json_dumps(val, JSON_NO_UTF8 | JSON_PRESERVE_ORDER);
+				s = json_dumps(val, JSON_NO_UTF8 | JSON_PRESERVE_ORDER | JSON_EOL);
-				fprintf(fp, "%s\n", s);
+				fprintf(fp, "%s", s);
 				dealloc(s);
 				json_decref(val);
 				fclose(fp);
@@ -3587,13 +3631,14 @@ static void *statsupdate(void *arg)
 			ghs = instance->dsps10080 * nonces;
 			suffix_string(ghs, suffix10080, 16, 0);
 
-			JSON_CPACK(val, "{ss,ss,ss,ss,ss,si}",
+			JSON_CPACK(val, "{ss,ss,ss,ss,ss,si,sf}",
 					"hashrate1m", suffix1,
 					"hashrate5m", suffix5,
 					"hashrate1hr", suffix60,
 					"hashrate1d", suffix1440,
 					"hashrate7d", suffix10080,
-					"workers", instance->workers);
+					"workers", instance->workers,
+					"bestshare", instance->best_diff);
 
 			snprintf(fname, 511, "%s/users/%s", ckp->logdir, instance->username);
 			fp = fopen(fname, "we");