1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
|
/* Research 046: the owner's side of the link. Takes the shim's cuts on a unix socket (u64le the cut's number, u64le
* its size, its ops), sends each over pure link as records of whole ops (ops.h: u8 1 on its last, the cut's number,
* the ops; at most RECORD bytes but for an op larger alone; SHIPPER_RECORD sets it), and answers the shim with the cut's number once the
* link says the follower saved its last record (pure_link_delivered): the owner's sync waits for that.
*
* shipper SOCKET pure://127.0.0.1:PORT/store KEY FOLLOWER_PUBLIC_HEX at its end, one JSON line of counts */
#define _GNU_SOURCE
#include "link.h"
#include "seal.h"
#include "ops.h"
#include <errno.h>
#include <pthread.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <time.h>
#include <unistd.h>
static size_t RECORD = 1 << 20; /* SHIPPER_RECORD: smaller, so a cut is several records (one torn at a kill) */
static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t saved = PTHREAD_COND_INITIALIZER;
static void notify(void *context) {
(void)context;
pthread_mutex_lock(&lock);
pthread_cond_broadcast(&saved);
pthread_mutex_unlock(&lock);
}
static bool whole_read(int fd, void *bytes, size_t size) {
for (unsigned char *p = bytes; size;) {
ssize_t n = read(fd, p, size);
if (n <= 0) { if (n < 0 && errno == EINTR) continue; return false; }
p += n, size -= (size_t)n;
}
return true;
}
int main(int argc, char **argv) {
if (argc != 5) { fprintf(stderr, "usage: %s SOCKET URL KEY FOLLOWER_PUBLIC_HEX\n", argv[0]); return 2; }
if (getenv("SHIPPER_RECORD")) RECORD = (size_t)atol(getenv("SHIPPER_RECORD"));
char error[256], public_hex[2 * PURE_SEAL_KEY + 1];
unsigned char own_public[PURE_SEAL_KEY], peer[PURE_SEAL_KEY];
PureSealKey *key = pure_seal_key(argv[3], public_hex, error, sizeof(error));
if (!key || !pure_seal_peer(public_hex, own_public) || !pure_seal_peer(argv[4], peer)) { fprintf(stderr, "shipper: %s\n", error); return 1; }
PureLink *link = pure_link_open(argv[2], "owner", 1, key, own_public, error, sizeof(error));
if (!link) { fprintf(stderr, "shipper: %s\n", error); return 1; }
pure_link_peer(link, peer);
if (!pure_link_start(link, notify, NULL, error, sizeof(error))) { fprintf(stderr, "shipper: %s\n", error); return 1; }
struct sockaddr_un address = {.sun_family = AF_UNIX};
if (strlen(argv[1]) >= sizeof(address.sun_path)) return 1;
strcpy(address.sun_path, argv[1]);
unlink(argv[1]);
int listening = socket(AF_UNIX, SOCK_STREAM, 0);
if (listening < 0 || bind(listening, (struct sockaddr *)&address, sizeof(address)) || listen(listening, 1)) {
perror("shipper: socket");
return 1;
}
printf("{\"ready\":true}\n");
fflush(stdout);
int shim = accept(listening, NULL, NULL);
if (shim < 0) return 1;
uint64_t seq = 0, cuts = 0, bytes = 0, records = 0, waited_ns = 0;
unsigned char *cut = NULL, *record = malloc(RECORD + 64);
size_t room = 0;
for (;;) {
unsigned char head[16];
if (!whole_read(shim, head, sizeof(head))) break; /* the owner ended (or was killed) */
uint64_t number = get64(head), size = get64(head + 8);
if (size > room && !(cut = realloc(cut, room = size))) return 1;
if (!whole_read(shim, cut, size)) break;
/* whole ops into records */
for (size_t at = 0; at < size;) {
size_t length = 9;
size_t end = at;
while (end < size) {
size_t n = op_size(cut + end, size - end);
if (!n) { fprintf(stderr, "shipper: a torn op in cut %llu\n", (unsigned long long)number); return 1; }
if (end > at && length + n > RECORD) break;
length += n, end += n;
}
unsigned char *r = length <= RECORD + 64 ? record : malloc(length);
r[0] = end == size;
put64(r + 1, number);
memcpy(r + 9, cut + at, end - at);
if (!pure_link_queue(link, ++seq, r, length)) { fprintf(stderr, "shipper: not queued\n"); return 1; }
if (r != record) free(r);
++records, bytes += length;
at = end;
}
struct timespec a, b;
clock_gettime(CLOCK_MONOTONIC, &a);
pure_link_push(link);
pthread_mutex_lock(&lock);
while (pure_link_delivered(link) < seq) {
struct timespec until;
clock_gettime(CLOCK_REALTIME, &until);
until.tv_nsec += 10 * 1000000;
if (until.tv_nsec >= 1000000000) until.tv_sec += 1, until.tv_nsec -= 1000000000;
pthread_cond_timedwait(&saved, &lock, &until);
}
pthread_mutex_unlock(&lock);
clock_gettime(CLOCK_MONOTONIC, &b);
waited_ns += (uint64_t)((b.tv_sec - a.tv_sec) * 1000000000LL + (b.tv_nsec - a.tv_nsec));
++cuts;
unsigned char ack[8];
put64(ack, number);
if (write(shim, ack, sizeof(ack)) != sizeof(ack)) break;
}
PureLinkCounts counts;
pure_link_counts(link, &counts);
printf("{\"cuts\":%llu,\"records\":%llu,\"record_bytes\":%llu,\"waited_ms\":%.1f,\"packets\":%llu,\"resent\":%llu}\n",
(unsigned long long)cuts, (unsigned long long)records, (unsigned long long)bytes, waited_ns / 1e6,
(unsigned long long)counts.packets, (unsigned long long)counts.resent);
pure_link_close(link);
pure_seal_key_free(key);
return 0;
}
|
