markow
/
ckpool
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Use async send proc as a separate thread from workqueues

master
Con Kolivas 10 years ago
parent
08a997e190
commit
7de43b1c6e
5 changed files with 65 additions and 72 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 70
      src/ckpool.c
  2. 4
      src/ckpool.h
  3. 10
      src/connector.c
  4. 28
      src/generator.c
  5. 25
      src/stratifier.c

70
src/ckpool.c
Unescape View File

@ -575,14 +575,32 @@ out:
static void childsighandler(const int sig); static void childsighandler(const int sig);
/* Send a single message to a process instance when there will be no response, struct proc_message {
* closing the socket immediately. */ proc_instance_t *pi;
void _send_proc(proc_instance_t *pi, const char *msg, const char *file, const char *func, const int line) char *msg;
const char *file;
const char *func;
int line;
};
/* Send all one way messages asynchronously so we can wait till the receiving
* end closes the socket to ensure all messages are received but no deadlocks
* can occur with 2 processes waiting for each other's socket closure. */
void *async_send_proc(void *arg)
{ {
struct proc_message *pm = (struct proc_message *)arg;
proc_instance_t *pi = pm->pi;
char *msg = pm->msg;
const char *file = pm->file;
const char *func = pm->func;
int line = pm->line;
char *path = pi->us.path; char *path = pi->us.path;
bool ret = false; bool ret = false;
int sockd; int sockd;
pthread_detach(pthread_self());
if (unlikely(!path || !strlen(path))) { if (unlikely(!path || !strlen(path))) {
LOGERR("Attempted to send message %s to null path in send_proc", msg ? msg : ""); LOGERR("Attempted to send message %s to null path in send_proc", msg ? msg : "");
goto out; goto out;
@ -593,14 +611,16 @@ void _send_proc(proc_instance_t *pi, const char *msg, const char *file, const ch
} }
/* At startup the pid fields are not set up before some processes are /* At startup the pid fields are not set up before some processes are
* forked so they never inherit them. */ * forked so they never inherit them. */
if (unlikely(!pi->pid)) if (unlikely(!pi->pid)) {
pi->pid = get_proc_pid(pi); pi->pid = get_proc_pid(pi);
if (!pi->pid) { if (!pi->pid) {
LOGALERT("Attempting to send message %s to non existent process %s", msg, pi->processname); LOGALERT("Attempting to send message %s to non existent process %s", msg, pi->processname);
return; goto out_nofail;
}
} }
if (unlikely(kill_pid(pi->pid, 0))) { if (unlikely(kill_pid(pi->pid, 0))) {
LOGALERT("Attempting to send message %s to non existent process %s", msg, pi->processname); LOGALERT("Attempting to send message %s to non existent process %s pid %d",
msg, pi->processname, pi->pid);
goto out; goto out;
} }
sockd = open_unix_client(path); sockd = open_unix_client(path);
@ -620,41 +640,25 @@ out:
LOGERR("Failure in send_proc from %s %s:%d", file, func, line); LOGERR("Failure in send_proc from %s %s:%d", file, func, line);
childsighandler(15); childsighandler(15);
} }
} out_nofail:
free(msg);
struct proc_message {
proc_instance_t *pi;
char *msg;
const char *file;
const char *func;
int line;
};
static void asp_send(ckpool_t __maybe_unused *ckp, struct proc_message *pm)
{
_send_proc(pm->pi, pm->msg, pm->file, pm->func, pm->line);
free(pm->msg);
free(pm); free(pm);
return NULL;
} }
/* Fore sending asynchronous messages to another process, the sending process /* Send a single message to a process instance when there will be no response,
* must have ckwqs of its own, referenced in the ckpool structure */ * closing the socket immediately. */
void _async_send_proc(ckpool_t *ckp, proc_instance_t *pi, const char *msg, const char *file, const char *func, const int line) void _send_proc(proc_instance_t *pi, const char *msg, const char *file, const char *func, const int line)
{ {
struct proc_message *pm; struct proc_message *pm = ckalloc(sizeof(struct proc_message));
pthread_t pth;
if (unlikely(!ckp->ckwqs)) {
LOGALERT("Workqueues not set up in async_send_proc!");
_send_proc(pi, msg, file, func, line);
return;
}
pm = ckzalloc(sizeof(struct proc_message));
pm->pi = pi; pm->pi = pi;
pm->msg = strdup(msg); pm->msg = strdup(msg);
pm->file = file; pm->file = file;
pm->func = func; pm->func = func;
pm->line = line; pm->line = line;
ckwq_add(ckp->ckwqs, &asp_send, pm); create_pthread(&pth, async_send_proc, pm);
} }
/* Send a single message to a process instance and retrieve the response, then /* Send a single message to a process instance and retrieve the response, then

4
src/ckpool.h
Unescape View File

@ -210,8 +210,6 @@ struct ckpool_instance {
/* Private data for each process */ /* Private data for each process */
void *data; void *data;
/* Private generic workqueues if this process has them */
ckwq_t *ckwqs;
}; };
#ifdef USE_CKDB #ifdef USE_CKDB
@ -236,8 +234,6 @@ void empty_buffer(connsock_t *cs);
int read_socket_line(connsock_t *cs, const int timeout); int read_socket_line(connsock_t *cs, const int timeout);
void _send_proc(proc_instance_t *pi, const char *msg, const char *file, const char *func, const int line); void _send_proc(proc_instance_t *pi, const char *msg, const char *file, const char *func, const int line);
#define send_proc(pi, msg) _send_proc(pi, msg, __FILE__, __func__, __LINE__) #define send_proc(pi, msg) _send_proc(pi, msg, __FILE__, __func__, __LINE__)
void _async_send_proc(ckpool_t *ckp, proc_instance_t *pi, const char *msg, const char *file, const char *func, const int line);
#define async_send_proc(ckp, pi, msg) _async_send_proc(ckp, pi, msg, __FILE__, __func__, __LINE__)
char *_send_recv_proc(proc_instance_t *pi, const char *msg, const char *file, const char *func, const int line); char *_send_recv_proc(proc_instance_t *pi, const char *msg, const char *file, const char *func, const int line);
#define send_recv_proc(pi, msg) _send_recv_proc(pi, msg, __FILE__, __func__, __LINE__) #define send_recv_proc(pi, msg) _send_recv_proc(pi, msg, __FILE__, __func__, __LINE__)
char *_send_recv_ckdb(const ckpool_t *ckp, const char *msg, const char *file, const char *func, const int line); char *_send_recv_ckdb(const ckpool_t *ckp, const char *msg, const char *file, const char *func, const int line);

10
src/connector.c
Unescape View File

@ -97,8 +97,6 @@ struct connector_data {
/* For protecting the pending sends list */ /* For protecting the pending sends list */
mutex_t sender_lock; mutex_t sender_lock;
pthread_cond_t sender_cond; pthread_cond_t sender_cond;
ckwq_t *ckwqs;
}; };
typedef struct connector_data cdata_t; typedef struct connector_data cdata_t;
@ -244,7 +242,7 @@ static void stratifier_drop_client(ckpool_t *ckp, const int64_t id)
char buf[256]; char buf[256];
sprintf(buf, "dropclient=%"PRId64, id); sprintf(buf, "dropclient=%"PRId64, id);
async_send_proc(ckp, ckp->stratifier, buf); send_proc(ckp->stratifier, buf);
} }
/* Invalidate this instance. Remove them from the hashtables we look up /* Invalidate this instance. Remove them from the hashtables we look up
@ -362,9 +360,9 @@ reparse:
* filtered by the stratifier. */ * filtered by the stratifier. */
if (likely(client->fd != -1)) { if (likely(client->fd != -1)) {
if (ckp->passthrough) if (ckp->passthrough)
async_send_proc(ckp, ckp->generator, s); send_proc(ckp->generator, s);
else else
async_send_proc(ckp, ckp->stratifier, s); send_proc(ckp->stratifier, s);
} }
free(s); free(s);
@ -870,8 +868,6 @@ int connector(proc_instance_t *pi)
LOGWARNING("%s connector starting", ckp->name); LOGWARNING("%s connector starting", ckp->name);
ckp->data = cdata; ckp->data = cdata;
cdata->ckp = ckp; cdata->ckp = ckp;
/* Connector only requires one work queue */
ckp->ckwqs = cdata->ckwqs = create_ckwqs(ckp, "conn", 1);
if (!ckp->serverurls) if (!ckp->serverurls)
cdata->serverfd = ckalloc(sizeof(int *)); cdata->serverfd = ckalloc(sizeof(int *));

28
src/generator.c
Unescape View File

@ -146,7 +146,6 @@ struct generator_data {
proxy_instance_t *dead_proxies; /* Disabled proxies */ proxy_instance_t *dead_proxies; /* Disabled proxies */
int proxy_notify_id; // Globally increasing notify id int proxy_notify_id; // Globally increasing notify id
ckmsgq_t *srvchk; // Server check message queue ckmsgq_t *srvchk; // Server check message queue
ckwq_t *ckwqs;
}; };
typedef struct generator_data gdata_t; typedef struct generator_data gdata_t;
@ -236,7 +235,7 @@ retry:
cs = &alive->cs; cs = &alive->cs;
LOGINFO("Connected to live server %s:%s", cs->url, cs->port); LOGINFO("Connected to live server %s:%s", cs->url, cs->port);
out: out:
async_send_proc(ckp, ckp->connector, alive ? "accept" : "reject"); send_proc(ckp->connector, alive ? "accept" : "reject");
return alive; return alive;
} }
@ -382,7 +381,7 @@ retry:
ret = submit_block(cs, buf + 12 + 64 + 1); ret = submit_block(cs, buf + 12 + 64 + 1);
memset(buf + 12 + 64, 0, 1); memset(buf + 12 + 64, 0, 1);
sprintf(blockmsg, "%sblock:%s", ret ? "" : "no", buf + 12); sprintf(blockmsg, "%sblock:%s", ret ? "" : "no", buf + 12);
async_send_proc(ckp, ckp->stratifier, blockmsg); send_proc(ckp->stratifier, blockmsg);
} else if (cmdmatch(buf, "checkaddr:")) { } else if (cmdmatch(buf, "checkaddr:")) {
if (validate_address(cs, buf + 10)) if (validate_address(cs, buf + 10))
send_unix_msg(sockd, "true"); send_unix_msg(sockd, "true");
@ -945,7 +944,7 @@ static void send_stratifier_deadproxy(ckpool_t *ckp, const int id, const int sub
char buf[256]; char buf[256];
sprintf(buf, "deadproxy=%d:%d", id, subid); sprintf(buf, "deadproxy=%d:%d", id, subid);
async_send_proc(ckp, ckp->stratifier, buf); send_proc(ckp->stratifier, buf);
} }
/* Remove the subproxy from the proxi list and put it on the dead list. /* Remove the subproxy from the proxi list and put it on the dead list.
@ -1092,7 +1091,7 @@ static void send_diff(ckpool_t *ckp, proxy_instance_t *proxi)
json_decref(json_msg); json_decref(json_msg);
ASPRINTF(&buf, "diff=%s", msg); ASPRINTF(&buf, "diff=%s", msg);
free(msg); free(msg);
async_send_proc(ckp, ckp->stratifier, buf); send_proc(ckp->stratifier, buf);
free(buf); free(buf);
} }
@ -1120,7 +1119,7 @@ static void send_notify(ckpool_t *ckp, proxy_instance_t *proxi, notify_instance_
json_decref(json_msg); json_decref(json_msg);
ASPRINTF(&buf, "notify=%s", msg); ASPRINTF(&buf, "notify=%s", msg);
free(msg); free(msg);
async_send_proc(ckp, ckp->stratifier, buf); send_proc(ckp->stratifier, buf);
free(buf); free(buf);
/* Send diff now as stratifier will not accept diff till it has a /* Send diff now as stratifier will not accept diff till it has a
@ -1310,7 +1309,7 @@ static void send_subscribe(ckpool_t *ckp, proxy_instance_t *proxi)
json_decref(json_msg); json_decref(json_msg);
ASPRINTF(&buf, "subscribe=%s", msg); ASPRINTF(&buf, "subscribe=%s", msg);
free(msg); free(msg);
async_send_proc(ckp, ckp->stratifier, buf); send_proc(ckp->stratifier, buf);
free(buf); free(buf);
} }
@ -1356,7 +1355,7 @@ static void stratifier_reconnect_client(ckpool_t *ckp, int64_t id)
char buf[256]; char buf[256];
sprintf(buf, "reconnclient=%"PRId64, id); sprintf(buf, "reconnclient=%"PRId64, id);
async_send_proc(ckp, ckp->stratifier, buf); send_proc(ckp->stratifier, buf);
} }
static void submit_share(gdata_t *gdata, json_t *val) static void submit_share(gdata_t *gdata, json_t *val)
@ -1730,9 +1729,9 @@ static void reconnect_proxy(proxy_instance_t *proxi)
create_pthread(&pth, proxy_reconnect, proxi); create_pthread(&pth, proxy_reconnect, proxi);
} }
static void reconnect_generator(ckpool_t *ckp) static void reconnect_generator(const ckpool_t *ckp)
{ {
async_send_proc(ckp, ckp->generator, "reconnect"); send_proc(ckp->generator, "reconnect");
} }
/* For receiving messages from an upstream pool to pass downstream. Responsible /* For receiving messages from an upstream pool to pass downstream. Responsible
@ -1789,7 +1788,7 @@ static void *passthrough_recv(void *arg)
/* Simply forward the message on, as is, to the connector to /* Simply forward the message on, as is, to the connector to
* process. Possibly parse parameters sent by upstream pool * process. Possibly parse parameters sent by upstream pool
* here */ * here */
async_send_proc(ckp, ckp->connector, cs->buf); send_proc(ckp->connector, cs->buf);
} }
return NULL; return NULL;
} }
@ -1994,10 +1993,10 @@ static proxy_instance_t *wait_best_proxy(ckpool_t *ckp, gdata_t *gdata)
if (ret) if (ret)
break; break;
async_send_proc(ckp, ckp->connector, "reject"); send_proc(ckp->connector, "reject");
sleep(1); sleep(1);
} }
async_send_proc(ckp, ckp->connector, ret ? "accept" : "reject"); send_proc(ckp->connector, ret ? "accept" : "reject");
return ret; return ret;
} }
@ -2026,7 +2025,7 @@ reconnect:
proxi->id, cs->url, cs->port); proxi->id, cs->url, cs->port);
dealloc(buf); dealloc(buf);
ASPRINTF(&buf, "proxy=%d", proxi->id); ASPRINTF(&buf, "proxy=%d", proxi->id);
async_send_proc(ckp, ckp->stratifier, buf); send_proc(ckp->stratifier, buf);
} }
} }
retry: retry:
@ -2216,7 +2215,6 @@ int generator(proc_instance_t *pi)
gdata = ckzalloc(sizeof(gdata_t)); gdata = ckzalloc(sizeof(gdata_t));
ckp->data = gdata; ckp->data = gdata;
gdata->ckp = ckp; gdata->ckp = ckp;
ckp->ckwqs = gdata->ckwqs = create_ckwqs(ckp, "gen", 1);
if (ckp->proxy) { if (ckp->proxy) {
char *buf = NULL; char *buf = NULL;

25
src/stratifier.c
Unescape View File

@ -831,7 +831,7 @@ static char *send_recv_generator(ckpool_t *ckp, const char *msg, const int prio)
return buf; return buf;
} }
static void send_generator(ckpool_t *ckp, const char *msg, const int prio) static void send_generator(const ckpool_t *ckp, const char *msg, const int prio)
{ {
sdata_t *sdata = ckp->data; sdata_t *sdata = ckp->data;
bool set; bool set;
@ -841,7 +841,7 @@ static void send_generator(ckpool_t *ckp, const char *msg, const int prio)
set = true; set = true;
} else } else
set = false; set = false;
async_send_proc(ckp, ckp->generator, msg); send_proc(ckp->generator, msg);
if (set) if (set)
sdata->gen_priority = 0; sdata->gen_priority = 0;
} }
@ -966,7 +966,7 @@ static void connector_drop_client(ckpool_t *ckp, const int64_t id)
LOGDEBUG("Stratifier requesting connector drop client %"PRId64, id); LOGDEBUG("Stratifier requesting connector drop client %"PRId64, id);
snprintf(buf, 255, "dropclient=%"PRId64, id); snprintf(buf, 255, "dropclient=%"PRId64, id);
async_send_proc(ckp, ckp->connector, buf); send_proc(ckp->connector, buf);
} }
static void drop_allclients(ckpool_t *ckp) static void drop_allclients(ckpool_t *ckp)
@ -1012,8 +1012,8 @@ static sdata_t *duplicate_sdata(const sdata_t *sdata)
memcpy(dsdata->donkeytxnbin, sdata->donkeytxnbin, 25); memcpy(dsdata->donkeytxnbin, sdata->donkeytxnbin, 25);
/* Use the same work queues for all subproxies */ /* Use the same work queues for all subproxies */
dsdata->ckwqs = sdata->ckwqs;
dsdata->ssends = sdata->ssends; dsdata->ssends = sdata->ssends;
dsdata->ckwqs = sdata->ckwqs;
dsdata->ckdbq = sdata->ckdbq; dsdata->ckdbq = sdata->ckdbq;
dsdata->sauthq = sdata->sauthq; dsdata->sauthq = sdata->sauthq;
@ -1139,7 +1139,7 @@ out_unlock:
static void reconnect_client(sdata_t *sdata, stratum_instance_t *client); static void reconnect_client(sdata_t *sdata, stratum_instance_t *client);
static void generator_recruit(ckpool_t *ckp) static void generator_recruit(const ckpool_t *ckp)
{ {
LOGINFO("Stratifer requesting more proxies from generator"); LOGINFO("Stratifer requesting more proxies from generator");
send_generator(ckp, "recruit", GEN_PRIORITY); send_generator(ckp, "recruit", GEN_PRIORITY);
@ -1780,7 +1780,7 @@ static void connector_test_client(ckpool_t *ckp, const int64_t id)
LOGDEBUG("Stratifier requesting connector test client %"PRId64, id); LOGDEBUG("Stratifier requesting connector test client %"PRId64, id);
snprintf(buf, 255, "testclient=%"PRId64, id); snprintf(buf, 255, "testclient=%"PRId64, id);
async_send_proc(ckp, ckp->connector, buf); send_proc(ckp->connector, buf);
} }
/* For creating a list of sends without locking that can then be concatenated /* For creating a list of sends without locking that can then be concatenated
@ -2567,7 +2567,7 @@ static void stratum_send_message(sdata_t *sdata, const stratum_instance_t *clien
* in proxy mode where we find a subproxy based on the current proxy with room * in proxy mode where we find a subproxy based on the current proxy with room
* for more clients. Signal the generator to recruit more subproxies if we are * for more clients. Signal the generator to recruit more subproxies if we are
* running out of room. */ * running out of room. */
static sdata_t *select_sdata(ckpool_t *ckp, sdata_t *ckp_sdata) static sdata_t *select_sdata(const ckpool_t *ckp, sdata_t *ckp_sdata)
{ {
proxy_t *current, *proxy, *subproxy, *best = NULL, *tmp, *tmpsub; proxy_t *current, *proxy, *subproxy, *best = NULL, *tmp, *tmpsub;
int best_id, best_subid = 0; int best_id, best_subid = 0;
@ -3968,7 +3968,6 @@ static void send_transactions(ckpool_t *ckp, json_params_t *jp);
static void parse_method(sdata_t *sdata, stratum_instance_t *client, const int64_t client_id, static void parse_method(sdata_t *sdata, stratum_instance_t *client, const int64_t client_id,
json_t *id_val, json_t *method_val, json_t *params_val, const char *address) json_t *id_val, json_t *method_val, json_t *params_val, const char *address)
{ {
ckpool_t *ckp = client->ckp;
const char *method; const char *method;
/* Random broken clients send something not an integer as the id so we /* Random broken clients send something not an integer as the id so we
@ -4015,14 +4014,14 @@ static void parse_method(sdata_t *sdata, stratum_instance_t *client, const int64
* to it since it's unauthorised. Set the flag just in case. */ * to it since it's unauthorised. Set the flag just in case. */
client->authorised = false; client->authorised = false;
snprintf(buf, 255, "passthrough=%"PRId64, client_id); snprintf(buf, 255, "passthrough=%"PRId64, client_id);
async_send_proc(ckp, ckp->connector, buf); send_proc(client->ckp->connector, buf);
return; return;
} }
/* We should only accept subscribed requests from here on */ /* We should only accept subscribed requests from here on */
if (!client->subscribed) { if (!client->subscribed) {
LOGINFO("Dropping unsubscribed client %"PRId64, client_id); LOGINFO("Dropping unsubscribed client %"PRId64, client_id);
connector_drop_client(ckp, client_id); connector_drop_client(client->ckp, client_id);
return; return;
} }
@ -4044,7 +4043,7 @@ static void parse_method(sdata_t *sdata, stratum_instance_t *client, const int64
* stratifier process to restart since it will have lost all * stratifier process to restart since it will have lost all
* the stratum instance data. Clients will just reconnect. */ * the stratum instance data. Clients will just reconnect. */
LOGINFO("Dropping unauthorised client %"PRId64, client_id); LOGINFO("Dropping unauthorised client %"PRId64, client_id);
connector_drop_client(ckp, client_id); connector_drop_client(client->ckp, client_id);
return; return;
} }
@ -4218,7 +4217,7 @@ static void ssend_process(ckpool_t *ckp, smsg_t *msg)
* connector process to be delivered */ * connector process to be delivered */
json_object_set_new_nocheck(msg->json_msg, "client_id", json_integer(msg->client_id)); json_object_set_new_nocheck(msg->json_msg, "client_id", json_integer(msg->client_id));
s = json_dumps(msg->json_msg, 0); s = json_dumps(msg->json_msg, 0);
async_send_proc(ckp, ckp->connector, s); send_proc(ckp->connector, s);
free(s); free(s);
free_smsg(msg); free_smsg(msg);
} }
@ -5081,7 +5080,7 @@ int stratifier(proc_instance_t *pi)
sdata->ssends = create_ckmsgq(ckp, "ssender", &ssend_process); sdata->ssends = create_ckmsgq(ckp, "ssender", &ssend_process);
/* Create as many generic workqueue threads as there are CPUs */ /* Create as many generic workqueue threads as there are CPUs */
threads = sysconf(_SC_NPROCESSORS_ONLN); threads = sysconf(_SC_NPROCESSORS_ONLN);
ckp->ckwqs = sdata->ckwqs = create_ckwqs(ckp, "strat", threads); sdata->ckwqs = create_ckwqs(ckp, "strat", threads);
sdata->sauthq = create_ckmsgq(ckp, "authoriser", &sauth_process); sdata->sauthq = create_ckmsgq(ckp, "authoriser", &sauth_process);
if (!CKP_STANDALONE(ckp)) { if (!CKP_STANDALONE(ckp)) {
sdata->ckdbq = create_ckmsgq(ckp, "ckdbqueue", &ckdbq_process); sdata->ckdbq = create_ckmsgq(ckp, "ckdbqueue", &ckdbq_process);

Write Preview
Loading…
Cancel
Save