Skip to content

Commit 7348623

Browse files
kvapkvap
committed
Add a DTMD benchmark.
1 parent aad5eae commit 7348623

File tree

5 files changed

+206
-28
lines changed

5 files changed

+206
-28
lines changed

‎contrib/pg_xtm/dtmd/include/util.h

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,10 +16,15 @@ char *join_path(const char *dir, const char *file);
1616
bool inrange(xid_t min, xid_t x, xid_t max);
1717
int falloc(int fd, off64_t size);
1818

19+
#if 0
20+
#define shout(...)
21+
#else
1922
#define shout(...) \
2023
do { \
2124
fprintf(stderr, __VA_ARGS__); \
2225
fflush(stderr); \
2326
} while (0)
2427

2528
#endif
29+
30+
#endif

‎contrib/pg_xtm/dtmd/src/main.c

Lines changed: 44 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
#include <string.h>
44
#include <unistd.h>
55
#include <assert.h>
6+
#include <time.h>
67

78
#include "clog.h"
89
#include "parser.h"
@@ -44,11 +45,11 @@ static void free_client_data(client_data_t *cd) {
4445
int next_client_id = 0;
4546
static void onconnect(void **client) {
4647
*client = create_client_data(next_client_id++);
47-
shout("[%d] connected\n", CLIENT_ID(*client));
48+
//shout("[%d] connected\n", CLIENT_ID(*client));
4849
}
4950

5051
static void ondisconnect(void *client) {
51-
shout("[%d] disconnected\n", CLIENT_ID(client));
52+
//shout("[%d] disconnected\n", CLIENT_ID(client));
5253
free_client_data(client);
5354
}
5455

@@ -193,10 +194,10 @@ static char *onvote(void *client, cmd_t *cmd, int vote) {
193194
switch (global_transaction_status(transactions + i)) {
194195
case NEGATIVE:
195196
if (global_transaction_mark(clg, transactions + i, NEGATIVE)) {
196-
shout(
197-
"[%d] VOTE: global transaction aborted\n",
198-
CLIENT_ID(client)
199-
);
197+
//shout(
198+
// "[%d] VOTE: global transaction aborted\n",
199+
// CLIENT_ID(client)
200+
//);
200201
transactions[i] = transactions[transactions_count - 1];
201202
transactions_count--;
202203
return strdup("+");
@@ -209,14 +210,14 @@ static char *onvote(void *client, cmd_t *cmd, int vote) {
209210
return strdup("-");
210211
}
211212
case NEUTRAL:
212-
shout("[%d] VOTE: vote counted\n", CLIENT_ID(client));
213+
//shout("[%d] VOTE: vote counted\n", CLIENT_ID(client));
213214
return strdup("+");
214215
case POSITIVE:
215216
if (global_transaction_mark(clg, transactions + i, POSITIVE)) {
216-
shout(
217-
"[%d] VOTE: global transaction committed\n",
218-
CLIENT_ID(client)
219-
);
217+
//shout(
218+
// "[%d] VOTE: global transaction committed\n",
219+
// CLIENT_ID(client)
220+
//);
220221
transactions[i] = transactions[transactions_count - 1];
221222
transactions_count--;
222223
return strdup("+");
@@ -351,26 +352,44 @@ static char *onnoise(void *client, cmd_t *cmd) {
351352
return strdup("-");
352353
}
353354

355+
static float now_s() {
356+
// current time in seconds
357+
struct timespec t;
358+
if (clock_gettime(CLOCK_MONOTONIC, &t) == 0) {
359+
return t.tv_sec + t.tv_nsec * 1e-9;
360+
} else {
361+
printf("Error while clock_gettime()\n");
362+
exit(0);
363+
}
364+
}
365+
354366
static char *oncmd(void *client, cmd_t *cmd) {
355-
shout_cmd(client, cmd);
367+
//shout_cmd(client, cmd);
356368

369+
float started = now_s();
370+
char *result = NULL;
357371
switch (cmd->cmd) {
358372
case CMD_BEGIN:
359-
return onbegin(client, cmd);
373+
result = onbegin(client, cmd);
374+
break;
360375
case CMD_COMMIT:
361-
return oncommit(client, cmd);
376+
result = oncommit(client, cmd);
377+
break;
362378
case CMD_ABORT:
363-
return onabort(client, cmd);
379+
result = onabort(client, cmd);
380+
break;
364381
case CMD_SNAPSHOT:
365-
return onsnapshot(client, cmd);
382+
result = onsnapshot(client, cmd);
383+
break;
366384
case CMD_STATUS:
367-
return onstatus(client, cmd);
385+
result = onstatus(client, cmd);
386+
break;
368387
default:
369388
return onnoise(client, cmd);
370389
}
371-
372-
assert(false); // the switch has holes in it?
373-
return NULL;
390+
float elapsed = now_s() - started;
391+
shout("cmd '%c' processed in %0.4f sec\n", cmd->cmd, elapsed);
392+
return result;
374393
}
375394

376395
char *destructive_concat(char *a, char *b) {
@@ -400,11 +419,11 @@ char *ondata(void *client, size_t len, char *data) {
400419
parser_t parser = CLIENT_PARSER(client);
401420
char *response = NULL;
402421

403-
shout(
404-
"[%d] got some data[%lu] %s\n",
405-
CLIENT_ID(client),
406-
len, data
407-
);
422+
//shout(
423+
// "[%d] got some data[%lu] %s\n",
424+
// CLIENT_ID(client),
425+
// len, data
426+
//);
408427

409428
// The idea is to feed each character through
410429
// the parser, which will return a cmd from

‎contrib/pg_xtm/libdtm.c

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,6 @@ typedef struct DTMConnData {
2020

2121
// Returns true if the write was successful.
2222
static bool dtm_write_char(DTMConn dtm, char c) {
23-
printf("writing %c\n", c);
2423
return write(dtm->sock, &c, 1) == 1;
2524
}
2625

@@ -56,7 +55,6 @@ static bool dtm_write_hex16(DTMConn dtm, xid_t i) {
5655
if (snprintf(buf, 17, "%016llx", i) != 16) {
5756
return false;
5857
}
59-
printf("writing %s\n", buf);
6058
return write(dtm->sock, buf, 16) == 16;
6159
}
6260

‎contrib/pg_xtm/libdtm/Makefile

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,10 @@ CC=clang -DTEST
22
CFLAGS=-g -Wall -I"../../../src/include"
33

44

5-
all: lib/libdtm.a bin/example
5+
all: lib/libdtm.a bin/example bin/bench
6+
7+
bin/bench: bindir lib/libdtm.a src/bench.c
8+
$(CC) $(CFLAGS) -o bin/bench -Iinclude -Llib src/bench.c -ldtm
69

710
bin/example: bindir lib/libdtm.a src/example.c
811
$(CC) $(CFLAGS) -o bin/example -Iinclude -Llib src/example.c -ldtm

‎contrib/pg_xtm/libdtm/src/bench.c

Lines changed: 153 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,153 @@
1+
#include <time.h>
2+
#include <stdio.h>
3+
#include <stdlib.h>
4+
5+
#include "libdtm.h"
6+
7+
#define NODES 1
8+
9+
#ifdef WIN32
10+
#include <windows.h>
11+
static float li2f(LARGE_INTEGER x) {
12+
float result = ((float)x.HighPart) * 4.294967296E9 + (float)((x).LowPart);
13+
return result;
14+
}
15+
16+
static float now_s() {
17+
LARGE_INTEGER freq, count;
18+
QueryPerformanceFrequency(&freq);
19+
QueryPerformanceCounter(&count);
20+
return li2f(count) / li2f(freq);
21+
}
22+
#else
23+
static float now_s() {
24+
// current time in seconds
25+
struct timespec t;
26+
if (clock_gettime(CLOCK_MONOTONIC, &t) == 0) {
27+
return t.tv_sec + t.tv_nsec * 1e-9;
28+
} else {
29+
printf("Error while clock_gettime()\n");
30+
exit(0);
31+
}
32+
}
33+
#endif
34+
35+
static void start_transaction(DTMConn conn, TransactionId base) {
36+
GlobalTransactionId gtid;
37+
gtid.nNodes = NODES;
38+
gtid.xids = malloc(sizeof(TransactionId) * gtid.nNodes);
39+
gtid.nodes = malloc(sizeof(NodeId) * gtid.nNodes);
40+
41+
int n;
42+
for (n = 0; n < gtid.nNodes; n++) {
43+
gtid.xids[n] = base + n;
44+
gtid.nodes[n] = n;
45+
}
46+
47+
if (!DtmGlobalStartTransaction(conn, &gtid)) {
48+
fprintf(stdout, "global transaction not started\n");
49+
exit(EXIT_FAILURE);
50+
}
51+
//fprintf(stdout, "global transaction started\n");
52+
53+
free(gtid.xids);
54+
free(gtid.nodes);
55+
}
56+
57+
static void commit_transaction(DTMConn conn, TransactionId base) {
58+
int n;
59+
for (n = 0; n < NODES; n++) {
60+
if (!DtmGlobalSetTransStatus(conn, n, base + n, TRANSACTION_STATUS_COMMITTED)) {
61+
fprintf(stdout, "global transaction not committed\n");
62+
exit(EXIT_FAILURE);
63+
}
64+
}
65+
//fprintf(stdout, "global transaction committed\n");
66+
}
67+
68+
static void abort_transaction(DTMConn conn, TransactionId base) {
69+
if (!DtmGlobalSetTransStatus(conn, 0, base + 0, TRANSACTION_STATUS_ABORTED)) {
70+
fprintf(stdout, "global transaction not aborted\n");
71+
exit(EXIT_FAILURE);
72+
}
73+
//fprintf(stdout, "global transaction aborted\n");
74+
}
75+
76+
static void show_snapshots(DTMConn conn, TransactionId base) {
77+
int i, n;
78+
for (n = 0; n < NODES; n++) {
79+
Snapshot s = malloc(sizeof(SnapshotData));
80+
s->xip = NULL;
81+
82+
if (!DtmGlobalGetSnapshot(conn, n, base + n, s)) {
83+
fprintf(stdout, "failed to get a snapshot[%d]\n", n);
84+
exit(EXIT_FAILURE);
85+
}
86+
//fprintf(stdout, "snapshot[%d, %#x]: xmin = %#x, xmax = %#x, active =", n, base + n, s->xmin, s->xmax);
87+
//for (i = 0; i < s->xcnt; i++) {
88+
// fprintf(stdout, " %#x", s->xip[i]);
89+
//}
90+
//fprintf(stdout, "\n");
91+
92+
free(s->xip);
93+
free(s);
94+
}
95+
}
96+
97+
static void show_status(DTMConn conn, TransactionId base) {
98+
int n;
99+
for (n = 0; n < NODES; n++) {
100+
XidStatus s = DtmGlobalGetTransStatus(conn, n, base + n);
101+
if (s == -1) {
102+
fprintf(stdout, "failed to get transaction status [%d, %#x]\n", n, base + n);
103+
exit(EXIT_FAILURE);
104+
}
105+
//fprintf(stdout, "status[%d, %#x]: ", n, base + n);
106+
//switch (s) {
107+
// case TRANSACTION_STATUS_COMMITTED:
108+
// fprintf(stdout, "committed\n");
109+
// break;
110+
// case TRANSACTION_STATUS_ABORTED:
111+
// fprintf(stdout, "aborted\n");
112+
// break;
113+
// case TRANSACTION_STATUS_IN_PROGRESS:
114+
// fprintf(stdout, "in progress\n");
115+
// break;
116+
// default:
117+
// fprintf(stdout, "(error)\n");
118+
// break;
119+
//}
120+
}
121+
}
122+
123+
int main(int argc, char **argv) {
124+
DTMConn conn = DtmConnect("localhost", 5431);
125+
if (!conn) {
126+
exit(1);
127+
}
128+
129+
TransactionId base = 42;
130+
131+
int transactions = atoi(argv[1]);
132+
int i;
133+
float started = now_s();
134+
for (i = 0; i < transactions; i++) {
135+
printf("-------- %d, base = %d -------- %0.6f sec \n", i, base, now_s() - started);
136+
start_transaction(conn, base);
137+
//show_snapshots(conn, base);
138+
//show_status(conn, base);
139+
commit_transaction(conn, base);
140+
base += NODES;
141+
}
142+
float elapsed = now_s() - started;
143+
144+
printf(
145+
"%d transactions in %0.2f sec -> %0.2f tps\n",
146+
transactions,
147+
elapsed,
148+
transactions / elapsed
149+
);
150+
151+
DtmDisconnect(conn);
152+
return EXIT_SUCCESS;
153+
}

0 commit comments

Comments
 (0)