diff options
Diffstat (limited to 'prototypes/connections')
| -rw-r--r-- | prototypes/connections/deliver.c | 191 | ||||
| -rw-r--r-- | prototypes/connections/deliver.h | 21 | ||||
| -rw-r--r-- | prototypes/connections/serve.c | 74 |
3 files changed, 255 insertions, 31 deletions
diff --git a/prototypes/connections/deliver.c b/prototypes/connections/deliver.c index c4136a2..0d3da64 100644 --- a/prototypes/connections/deliver.c +++ b/prototypes/connections/deliver.c @@ -1,5 +1,6 @@ #define _POSIX_C_SOURCE 200809L #include "deliver.h" +#include "message-json.h" #include "module-text.h" #include <errno.h> #include <netdb.h> @@ -18,10 +19,14 @@ #include <time.h> #include <unistd.h> -enum { TIMEOUT_S = 10, FIRST_PAUSE_MS = 100, LONGEST_PAUSE_MS = 10000, GROUPS = PURE_MODULE_NAME / 2, BATCH_MOST = 256 }; -enum { DELIVERED, AGAIN, REFUSED_BY_PEER, TLS_FAILED }; +enum { TIMEOUT_S = 10, FIRST_PAUSE_MS = 100, LONGEST_PAUSE_MS = 10000, GROUPS = PURE_MODULE_NAME / 2, BATCH_MOST = 256, + ANSWER_MOST = 1 << 20 }; +enum { DELIVERED, AGAIN, REFUSED_BY_PEER, TLS_FAILED, ANSWER_UNREAD }; -typedef struct { char *body; size_t size; uint64_t seq; } Pending; +typedef struct { char *body; size_t size; uint64_t seq; unsigned char *sent; size_t sent_size; } Pending; +/* An answer made into its input's message: a table of the fields' starts and + * sizes (u32le), then the payload from its prefix on. */ +typedef struct { uint64_t seq; unsigned char *bytes; size_t size, table; } Answer; struct PureWire { const PureModule *module; @@ -40,6 +45,11 @@ struct PureWire { int fd; /* the connection kept for the next send, or -1; changed under the lock */ bool tls; /* https: the connection is TLS */ SSL *ssl; /* its TLS session */ + int answer_input; /* the input answers become events of, or -1 */ + Answer *answers; /* made, for the owner to take in order */ + size_t answer_head, answer_count, answer_capacity; + Answer taken; /* the one the owner took last, kept until it takes the next */ + uint32_t placed[2 * PURE_MODULE_FIELDS]; }; /* ---- TLS: one client context, verifying the receiver's certificate against @@ -226,12 +236,42 @@ PureWire *pure_wire_open(const PureModule *m, const char *output, const char *ur memcpy(w->source, source, length + 1); w->fd = -1; w->tls = tls; + w->answer_input = -1; pthread_mutex_init(&w->lock, NULL); pthread_cond_init(&w->changed, NULL); return w; } uint32_t pure_wire_output(const PureWire *w) { return w->output; } + +bool pure_wire_reply(PureWire *w, uint32_t input, char *error, size_t size) { + const PureModule *m = w->module; + if (input >= m->inputs || m->input[input].kind != PURE_KIND_MESSAGE || m->input[input].keyed) { + snprintf(error, size, "an answer goes to a message input without a key"); + return false; + } + w->answer_input = (int)input; + return true; +} + +bool pure_wire_answer(PureWire *w, PureWireAnswer *out) { + free(w->taken.bytes); + w->taken = (Answer){0}; + pthread_mutex_lock(&w->lock); + bool some = w->answer_count > 0; + if (some) { + w->taken = w->answers[w->answer_head++]; + --w->answer_count; + w->delivered = w->taken.seq; /* trimmed after the owner applies it */ + } + pthread_mutex_unlock(&w->lock); + if (!some) return false; + const PureInputShape *in = &w->module->input[w->answer_input]; + for (size_t i = 0; i < 2 * (size_t)in->field_count; ++i) w->placed[i] = read_u32(w->taken.bytes + 4 * i); + *out = (PureWireAnswer){(uint32_t)w->answer_input, w->taken.seq, + {w->taken.bytes + w->taken.table, w->taken.size - w->taken.table}, w->placed, w->source}; + return true; +} uint64_t pure_wire_queued(const PureWire *w) { return w->queued; } uint64_t pure_wire_delivered(PureWire *w) { @@ -275,7 +315,14 @@ bool pure_wire_queue(PureWire *w, uint64_t seq, PureSpan payload) { w->capacity = capacity; } } - w->queue[w->head + w->count++] = (Pending){t.bytes, t.size, seq}; + unsigned char *sent = NULL; + if (w->answer_input >= 0 && !(sent = malloc(payload.size ? payload.size : 1))) { + pthread_mutex_unlock(&w->lock); + free(t.bytes); + return false; + } + if (sent && payload.size) memcpy(sent, payload.data, payload.size); + w->queue[w->head + w->count++] = (Pending){t.bytes, t.size, seq, sent, payload.size}; w->queued = seq; pthread_cond_signal(&w->changed); pthread_mutex_unlock(&w->lock); @@ -294,6 +341,13 @@ static bool fill(Reader *r) { if (w->ssl) n = SSL_read(w->ssl, r->bytes + r->end, (int)(sizeof(r->bytes) - r->end)); else do n = recv(w->fd, r->bytes + r->end, sizeof(r->bytes) - r->end, 0); while (n < 0 && errno == EINTR); if (n <= 0) return false; +#ifdef TCP_QUICKACK + /* Acknowledge at once: a server that writes an answer in pieces holds the + * next piece until the first is acknowledged (Nagle), which a delayed + * acknowledgement would stall for 40 ms. The flag does not stay set. */ + int one = 1; + setsockopt(w->fd, IPPROTO_TCP, TCP_QUICKACK, &one, sizeof(one)); +#endif r->end += (size_t)n; return true; } @@ -320,14 +374,17 @@ static bool read_line(Reader *r, char *line, size_t size) { } } -static bool skip(Reader *r, uint64_t size) { +/* size bytes of the body: kept in into (up to ANSWER_MOST), or skipped when it is NULL. */ +static bool take(Reader *r, uint64_t size, Text *into) { + if (into && size > ANSWER_MOST - into->size) return false; while (size) { if (r->at == r->end && !fill(r)) return false; size_t n = r->end - r->at < size ? r->end - r->at : (size_t)size; + if (into) put(into, r->bytes + r->at, n); r->at += n; size -= n; } - return true; + return !into || !into->failed; } static bool send_all(PureWire *w, const char *bytes, size_t size) { @@ -397,7 +454,7 @@ static bool refusal(const char *error) { /* The answer to one request: its status, whether a Pure receiver refused it, * and whether the connection may carry the next. */ -static bool read_answer(PureWire *w, int *status, bool *refused, bool *keep, unsigned *batch) { +static bool read_answer(PureWire *w, int *status, bool *refused, bool *keep, unsigned *batch, Text *body) { Reader r = {.wire = w}; char line[1024]; if (!read_line(&r, line, sizeof(line)) || strncmp(line, "HTTP/1.", 7) || strlen(line) < 12) return false; @@ -414,15 +471,18 @@ static bool read_answer(PureWire *w, int *status, bool *refused, bool *keep, uns if (!read_line(&r, line, sizeof(line))) return false; chunk = strtoull(line, NULL, 16); if (!chunk) break; - if (!skip(&r, chunk) || !read_line(&r, line, sizeof(line))) return false; + if (!take(&r, chunk, body) || !read_line(&r, line, sizeof(line))) return false; } while (read_line(&r, line, sizeof(line)) && *line) {} return true; } if (*status == 204 || *status == 304 || *status < 200) return true; - if (sized) return skip(&r, length); + if (sized) return take(&r, length, body); *keep = false; /* the body runs to the close */ - while (fill(&r)) r.at = r.end; + while (fill(&r)) { + if (body && !take(&r, r.end - r.at, body)) return false; + r.at = r.end; + } return true; } char *value = strchr(line, ':'); @@ -438,9 +498,64 @@ static bool read_answer(PureWire *w, int *status, bool *refused, bool *keep, uns return false; } +static void write_u32(unsigned char *p, uint32_t v) { for (unsigned i = 0; i < 4; ++i) p[i] = (unsigned char)(v >> (8 * i)); } + +/* The answer (status, body) to the send p as a message of the answer input: + * fields from the body's JSON object by name, others skipped; a field the + * body lacks from the send's field of that name and type; a u32 field named + * http_status from the status. False when a field has none of these. */ +static bool form_answer(const PureWire *w, int status, const Text *body, const Pending *p, Answer *out) { + const PureModule *m = w->module; + const PureInputShape *in = &m->input[w->answer_input], *sent = &m->input[w->output]; + size_t table = 8 * (size_t)in->field_count, total = table + in->prefix + body->size + p->sent_size; + unsigned char *bytes = calloc(1, total ? total : 1); + if (!bytes) return false; + unsigned char *prefix = bytes + table, *text = prefix + in->prefix, *copy = text + body->size; + if (body->size) memcpy(text, body->bytes, body->size); + if (p->sent_size) memcpy(copy, p->sent, p->sent_size); + PureJsonMessage message = {.module = m, .shape = in, .payload = prefix, .table = bytes, .lenient = true}; + size_t at = pure_json_space(text, 0, body->size); + if (at < body->size && text[at] == '{' && pure_json_message(&message, text, body->size, &at)) { + free(bytes); + return false; + } + for (uint32_t f = 0; f < in->field_count; ++f) { + if (message.seen[f]) continue; + const PureField *field = &m->field[in->field_start + f]; + if (field->type == PURE_FIELD_U32 && !strcmp(field->name, "http_status")) { + write_u32(prefix + field->offset, (uint32_t)status); + continue; + } + /* The send's field of the same name and type, and where its bytes are. */ + size_t start = sent->prefix, length = 0; + const PureField *from = NULL; + for (uint32_t g = 0; g < sent->field_count && !from; ++g) { + const PureField *other = &m->field[sent->field_start + g]; + bool variable = other->type == PURE_FIELD_BYTES || other->type == PURE_FIELD_TEXT; + if (variable) length = other->offset == PURE_FIELD_REST ? p->sent_size - start : read_u32(p->sent + other->offset); + if (!strcmp(other->name, field->name) && other->type == field->type) from = other; + else if (variable) start += length; + } + bool variable = field->type == PURE_FIELD_BYTES || field->type == PURE_FIELD_TEXT; + if (!from || (variable && (start > p->sent_size || length > p->sent_size - start))) { + free(bytes); + return false; + } + if (field->type == PURE_FIELD_U32 || field->type == PURE_FIELD_NUMBER) + memcpy(prefix + field->offset, p->sent + from->offset, field->type == PURE_FIELD_U32 ? 4 : 8); + else { + write_u32(bytes + 8 * f, (uint32_t)(copy + start - prefix)); + write_u32(bytes + 8 * f + 4, (uint32_t)length); + } + } + *out = (Answer){p->seq, bytes, total, table}; + return true; +} + /* One request for count sends from p (numbered in a row): one object, or - * an array of them for a receiver that takes batches. */ -static int post(PureWire *w, const Pending *p, unsigned count) { + * an array of them for a receiver that takes batches. A final answer to a + * send whose answers are wired is made into *answer. */ +static int post(PureWire *w, const Pending *p, unsigned count, Answer *answer) { size_t body = count > 1 ? count + 1 : 0; for (unsigned i = 0; i < count; ++i) body += p[i].size; char head[2048], batch[32] = ""; @@ -473,19 +588,30 @@ static int post(PureWire *w, const Pending *p, unsigned count) { int status; bool refused, keep, sent = send_all(w, head, (size_t)size); unsigned batch_size = 0; + Text body = {0}; for (unsigned i = 0; sent && i < count; ++i) sent = (!i || count == 1 || send_all(w, ",", 1)) && (i || count == 1 || send_all(w, "[", 1)) && send_all(w, p[i].body, p[i].size); if (sent && count > 1) sent = send_all(w, "]", 1); - if (!sent || !read_answer(w, &status, &refused, &keep, &batch_size)) { + if (!sent || !read_answer(w, &status, &refused, &keep, &batch_size, w->answer_input >= 0 ? &body : NULL)) { + free(body.bytes); drop_connection(w); if (fresh) return AGAIN; continue; } if (!keep) drop_connection(w); - if (batch_size) w->batch = batch_size; + if (batch_size && w->answer_input < 0) w->batch = batch_size; + if (status == 408 || status == 429 || status >= 500) { + free(body.bytes); + return AGAIN; + } + if (w->answer_input >= 0 && status >= 200 && status < 500) { /* a final answer: the app hears it */ + bool formed = form_answer(w, status, &body, p, answer); + free(body.bytes); + return formed ? DELIVERED : ANSWER_UNREAD; + } + free(body.bytes); if ((status >= 200 && status < 300) || (status >= 400 && status < 500 && refused)) return DELIVERED; - if (status == 408 || status == 429 || status >= 500) return AGAIN; return REFUSED_BY_PEER; } return AGAIN; @@ -507,13 +633,31 @@ static void *deliver(void *context) { memcpy(next, w->queue + w->head, count * sizeof(*next)); ++w->attempts; pthread_mutex_unlock(&w->lock); - int outcome = post(w, next, count); + Answer answer = {0}; + int outcome = post(w, next, count, &answer); pthread_mutex_lock(&w->lock); + if (outcome == DELIVERED && answer.bytes) { /* for the owner, who then counts it delivered */ + if (w->answer_head + w->answer_count == w->answer_capacity) { + if (w->answer_head) { + memmove(w->answers, w->answers + w->answer_head, w->answer_count * sizeof(*w->answers)); + w->answer_head = 0; + } else { + size_t capacity = w->answer_capacity ? 2 * w->answer_capacity : 16; + Answer *grown = realloc(w->answers, capacity * sizeof(*grown)); + if (!grown) { free(answer.bytes); outcome = AGAIN; } + else { w->answers = grown; w->answer_capacity = capacity; } + } + } + if (outcome == DELIVERED) w->answers[w->answer_head + w->answer_count++] = answer; + } if (outcome == DELIVERED) { - for (unsigned i = 0; i < count; ++i) free(w->queue[w->head + i].body); + for (unsigned i = 0; i < count; ++i) { + free(w->queue[w->head + i].body); + free(w->queue[w->head + i].sent); + } w->head += count; w->count -= count; - w->delivered = next[count - 1].seq; + if (!answer.bytes) w->delivered = next[count - 1].seq; w->deliveries += count; pause_ms = 0; pthread_mutex_unlock(&w->lock); @@ -521,9 +665,10 @@ static void *deliver(void *context) { pthread_mutex_lock(&w->lock); continue; } - if ((outcome == REFUSED_BY_PEER || outcome == TLS_FAILED) && reported != next[0].seq) { + if (outcome != AGAIN && reported != next[0].seq) { fprintf(stderr, "{\"wire\":\"%s\",\"seq\":%llu,\"error\":\"%s\"}\n", w->source, (unsigned long long)next[0].seq, outcome == TLS_FAILED ? "the receiver's certificate is not trusted; trying again" : + outcome == ANSWER_UNREAD ? "its answer does not make the answer input's message; trying again" : "the receiver does not take it; trying again"); reported = next[0].seq; } @@ -557,7 +702,13 @@ void pure_wire_close(PureWire *w) { if (w->started) pthread_join(w->thread, NULL); if (w->ssl) SSL_free(w->ssl); if (w->fd >= 0) close(w->fd); - for (size_t i = 0; i < w->count; ++i) free(w->queue[w->head + i].body); + for (size_t i = 0; i < w->count; ++i) { + free(w->queue[w->head + i].body); + free(w->queue[w->head + i].sent); + } + for (size_t i = 0; i < w->answer_count; ++i) free(w->answers[w->answer_head + i].bytes); + free(w->answers); + free(w->taken.bytes); free(w->queue); pthread_cond_destroy(&w->changed); pthread_mutex_destroy(&w->lock); diff --git a/prototypes/connections/deliver.h b/prototypes/connections/deliver.h index 13048fc..2d1da29 100644 --- a/prototypes/connections/deliver.h +++ b/prototypes/connections/deliver.h @@ -34,6 +34,27 @@ uint64_t pure_wire_delivered(PureWire *); bool pure_wire_start(PureWire *, void (*notify)(void *), void *context); /* Stop (a delivery under way is finished or abandoned) and free. */ void pure_wire_close(PureWire *); +/* Answers as events: the final answer to each send (a 2xx, or a 4xx but 408 + * and 429) becomes a message event of input (a message without a key), in + * the order sent: its fields come from the answer's JSON object by name + * (other members are skipped), a field the answer lacks from the send's + * field of that name and type (as a request's id), and a u32 field named + * http_status is the answer's status. An answer that makes no message is + * logged and the send tried again every 10 s. Answers go one per request. */ +bool pure_wire_reply(PureWire *, uint32_t input, char *error, size_t size); +/* The owner's side: the next answer in order, or false. Taking it counts + * its send delivered (the owner applies the answer, then trims); what out + * points to is the wire's until the next call. The answer is applied with + * pure_app_deliver from the wire's source with the send's number, so an + * answer that comes again after a restart is applied once. */ +typedef struct { + uint32_t input; + uint64_t seq; + PureSpan payload; + const uint32_t *placed; + const char *source; +} PureWireAnswer; +bool pure_wire_answer(PureWire *, PureWireAnswer *out); /* Counts for a host's report. */ void pure_wire_counts(PureWire *, uint64_t *delivered, uint64_t *attempts); #endif diff --git a/prototypes/connections/serve.c b/prototypes/connections/serve.c index 88a9a6f..60771e1 100644 --- a/prototypes/connections/serve.c +++ b/prototypes/connections/serve.c @@ -150,7 +150,8 @@ typedef struct { /* The owner's: outputs wired to HTTP receivers, and how far each was trimmed. */ PureWire *wires[PURE_MODULE_INPUTS]; unsigned wire_count; - uint64_t trimmed[PURE_MODULE_INPUTS]; + uint64_t trimmed[PURE_MODULE_INPUTS], answers; + bool held[PURE_MODULE_INPUTS]; /* an answer failed to apply: the wire's sends stay until a restart */ } Executor; /* A request made ready: its effect, what it names, and the thread that answers it. */ @@ -1051,14 +1052,32 @@ static void after_save(Executor *ex) { if (ex->wire_count) pure_app_outbox(&ex->app, queue_send, &q); } -/* What the wires delivered leaves the outbox (with the next save). */ -static void trim_delivered(Executor *ex) { +/* The answers the wires made become events, in order; then what the wires + * delivered leaves the outbox (with the next save). True when an answer was + * applied (the caller saves). */ +static bool take_deliveries(Executor *ex) { + bool answered = false; for (unsigned i = 0; i < ex->wire_count; ++i) { + PureWireAnswer answer; + while (pure_wire_answer(ex->wires[i], &answer)) { + ex->app.now = wall_ms(); + PureResult result = pure_app_deliver(&ex->app, answer.input, answer.source, answer.seq, answer.payload, + answer.placed); + if (result != PURE_OK && ex->app.fault == PURE_FAULT_NONE) { + ex->held[i] = true; /* not applied and not refused: sent again after a restart */ + fprintf(stderr, "{\"wire\":\"%s\",\"seq\":%" PRIu64 ",\"error\":\"its answer was not applied\"}\n", + answer.source, answer.seq); + } + pure_app_finish(&ex->app); + answered = true; + ++ex->answers; + } uint64_t delivered = pure_wire_delivered(ex->wires[i]); - if (delivered <= ex->trimmed[i]) continue; + if (delivered <= ex->trimmed[i] || ex->held[i]) continue; pure_app_trim(&ex->app, pure_wire_output(ex->wires[i]), delivered); ex->trimmed[i] = delivered; } + return answered; } /* One save for a burst: its requests are durable with it or fail with it. */ @@ -1290,8 +1309,14 @@ static void *run_owner(void *context) { owner->delivered = false; pthread_mutex_unlock(&owner->lock); long due = run_timers(&owner->executor); - if (delivered) trim_delivered(&owner->executor); + bool answered = delivered && take_deliveries(&owner->executor); pthread_mutex_lock(&owner->lock); + /* Answers are saved with the next round, or now if none waits. */ + if (answered && !owner->count && !owner->stopping) { + pthread_mutex_unlock(&owner->lock); + if (pure_app_save(&owner->executor.app) == PURE_OK) after_save(&owner->executor); + pthread_mutex_lock(&owner->lock); + } if (owner->count || owner->stopping) break; if (owner->delivered) continue; if (due < 0) { @@ -1604,6 +1629,30 @@ static bool open_wires(Executor *ex, const PureModule *m, const char *const *sen return true; } +/* --reply OUTPUT=INPUT: the answers to that output's sends become INPUT's events. */ +static bool wire_replies(Executor *ex, const PureModule *m, const char *const *replies, unsigned count) { + for (unsigned i = 0; i < count; ++i) { + const char *equals = strchr(replies[i], '='); + char output[PURE_MODULE_NAME], error[160] = ""; + size_t length = equals ? (size_t)(equals - replies[i]) : 0; + int input = equals ? pure_module_input(m, equals + 1) : -1; + PureWire *wire = NULL; + if (length && length < sizeof(output)) { + memcpy(output, replies[i], length); + output[length] = 0; + int index = pure_module_input(m, output); + for (unsigned k = 0; k < ex->wire_count && index >= 0; ++k) + if (pure_wire_output(ex->wires[k]) == (uint32_t)index) wire = ex->wires[k]; + } + if (!wire || input < 0 || !pure_wire_reply(wire, (uint32_t)input, error, sizeof(error))) { + fprintf(stderr, "--reply %s: %s\n", replies[i], !wire ? "takes OUTPUT=INPUT of an output given --send" : + input < 0 ? "no such input" : error); + return false; + } + } + return true; +} + /* --threads auto: a connection thread for each CPU this process may run on, * less the one the owner of an app with states takes. */ static unsigned auto_threads(bool owner) { @@ -1617,8 +1666,8 @@ static unsigned auto_threads(bool owner) { int main(int argc, char **argv) { static PureModule module; static Owner owner; - const char *module_path = NULL, *store = NULL, *name = NULL, *sends[PURE_MODULE_INPUTS]; - unsigned send_count = 0; + const char *module_path = NULL, *store = NULL, *name = NULL, *sends[PURE_MODULE_INPUTS], *replies[PURE_MODULE_INPUTS]; + unsigned send_count = 0, reply_count = 0; unsigned port = 8767, threads = 1; size_t budget = 64u << 20, capacity = 1024; uint64_t timeout_ms = 10000; @@ -1630,6 +1679,7 @@ int main(int argc, char **argv) { else if (!strcmp(argv[i], "--send") && send_count < PURE_MODULE_INPUTS) sends[send_count++] = argv[i + 1]; else if (!strcmp(argv[i], "--name")) name = argv[i + 1]; else if (!strcmp(argv[i], "--send-ca")) pure_wire_trust(argv[i + 1]); + else if (!strcmp(argv[i], "--reply") && reply_count < PURE_MODULE_INPUTS) replies[reply_count++] = argv[i + 1]; else if (!strcmp(argv[i], "--port") && number(argv[i + 1], 65535, &value)) port = (unsigned)value; else if (!strcmp(argv[i], "--budget") && number(argv[i + 1], SIZE_MAX, &value) && value) budget = (size_t)value; else if (!strcmp(argv[i], "--connections") && number(argv[i + 1], 1u << 20, &value) && value) capacity = (size_t)value; @@ -1671,7 +1721,8 @@ int main(int argc, char **argv) { return 1; owner.executor.reads_first = true; owner.executor.publishes = true; - if (!open_wires(&owner.executor, &module, sends, send_count, name)) return 2; + if (!open_wires(&owner.executor, &module, sends, send_count, name) || + !wire_replies(&owner.executor, &module, replies, reply_count)) return 2; after_save(&owner.executor); /* the snapshot, and the sends stored before */ for (unsigned i = 0; i < owner.executor.wire_count; ++i) if (!pure_wire_start(owner.executor.wires[i], wire_delivered, &owner)) { fputs("Cannot start a wire\n", stderr); return 1; } @@ -1731,15 +1782,16 @@ int main(int argc, char **argv) { fprintf(stderr, "{\"active\":%zu,\"retained_payload_bytes\":%zu,\"accepted\":%" PRIu64 ",\"requests\":%" PRIu64 ",\"bursts\":%" PRIu64 ",\"burst_events\":%" PRIu64 ",\"ticks\":%" PRIu64 ",\"expired\":%" PRIu64 ",\"threads\":%u,\"owner\":%s,\"snapshots\":%" PRIu64 ",\"views_from_snapshots\":%" PRIu64 - ",\"delivered\":%" PRIu64 ",\"delivery_attempts\":%" PRIu64 ",\"commit_groups\":%lu,\"grouped_commits\":%lu}\n", + ",\"delivered\":%" PRIu64 ",\"delivery_attempts\":%" PRIu64 ",\"answers\":%" PRIu64 + ",\"commit_groups\":%lu,\"grouped_commits\":%lu}\n", active, retained, accepted, requests, bursts, events, ticks, expired, threads, owned ? "true" : "false", - owner.executor.publications, shown, delivered, attempts, groups, grouped); + owner.executor.publications, shown, delivered, attempts, owner.executor.answers, groups, grouped); free(hosts); free(thread); return 0; usage: fputs("pure-serve --module FILE.pmod --store DIRECTORY [--port 8767] [--threads 1|N|auto] [--budget BYTES] " "[--connections 1024] [--timeout-ms 10000] [--send OUTPUT=http[s]://host:port/path ...] [--name NAME] " - "[--send-ca FILE]\n", stderr); + "[--send-ca FILE] [--reply OUTPUT=INPUT ...]\n", stderr); return 2; } |