Merge pull request #23 from sometimesfood/fix-compiler-warnings
[pachi.git] / distributed / distributed.c
blobff7ad79e9e789ec0477b14b82b6632e347c1b9ed
1 /* This is a master for the "distributed" engine. It receives connections
2 * from slave machines, sends them gtp commands, then aggregates the
3 * results. It can also act as a proxy for the logs of all slave machines.
4 * The slave machines must run with engine "uct" (not "distributed").
5 * The master sends pachi-genmoves gtp commands regularly to each slave,
6 * gets as replies a list of nodes, their number of playouts
7 * and their value. The master then picks the most popular move
8 * among the top level nodes. */
10 /* With time control, the master waits for all slaves, except
11 * when the allowed time is already passed. In this case the
12 * master picks among the available replies, or waits for just
13 * one reply if there is none yet.
14 * Without time control, the master waits until the desired
15 * number of games have been simulated. In this case the -t
16 * parameter for the master should be the sum of the parameters
17 * for all slaves. */
19 /* The master sends updated statistics for the best nodes in each
20 * genmoves command. They are incremental updates from all other
21 * slaves (so they exclude contributions from the target slave).
22 * The slaves reply with just their own stats. So both master and
23 * slave remember what was previously sent. A slave remembers in
24 * the tree ("pu" field), which is stable across moves. The slave
25 * also has a temporary hash table to map received coord paths
26 * to tree nodes; the hash table is cleared at each new move.
27 * The master remembers stats in a queue of received buffers that
28 * are merged together, plus one hash table per slave. The master
29 * queue and the hash tables are cleared at each new move. */
31 /* To allow the master to select the best move, slaves also send
32 * absolute playout counts for the best top level nodes (children
33 * of the root node), including contributions from other slaves.
34 * The master sums these counts and picks the best sum, which is
35 * equivalent to picking the best average. (The master cannot
36 * use the incremental stats sent in binary form because they
37 * are not maintained across moves, so playouts from previous
38 * moves would be lost.) */
40 /* The master-slave protocol has fault tolerance. If a slave is
41 * out of sync, the master sends it the appropriate command history. */
43 /* Pass me arguments like a=b,c=d,...
44 * Supported arguments:
45 * slave_port=SLAVE_PORT slaves connect to this port; this parameter is mandatory.
46 * max_slaves=MAX_SLAVES default 24
47 * shared_nodes=SHARED_NODES default 10K
48 * stats_hbits=STATS_HBITS default 21. 2^stats_bits = hash table size
49 * slaves_quit=0|1 quit gtp command also sent to slaves, default false.
50 * proxy_port=PROXY_PORT slaves optionally send their logs to this port.
51 * Warning: with proxy_port, the master stderr mixes the logs of all
52 * machines but you can separate them again:
53 * slave logs: sed -n '/< .*:/s/.*< /< /p' logfile
54 * master logs: perl -0777 -pe 's/<[ <].*:.*\n//g' logfile
57 /* A configuration without proxy would have one master run on masterhost as:
58 * pachi -e distributed slave_port=1234
59 * and N slaves running as:
60 * pachi -e uct -g masterhost:1234 slave
61 * With log proxy:
62 * pachi -e distributed slave_port=1234,proxy_port=1235
63 * pachi -e uct -g masterhost:1234 -l masterhost:1235 slave
64 * If the master itself runs on a machine other than that running gogui,
65 * gogui-twogtp, kgsGtp or cgosGtp, it can redirect its gtp port:
66 * pachi -e distributed -g 10000 slave_port=1234,proxy_port=1235
69 #include <assert.h>
70 #include <stdio.h>
71 #include <stdlib.h>
72 #include <string.h>
73 #include <time.h>
74 #include <unistd.h>
75 #include <sys/types.h>
77 #define DEBUG
79 #include "engine.h"
80 #include "move.h"
81 #include "timeinfo.h"
82 #include "playout.h"
83 #include "stats.h"
84 #include "mq.h"
85 #include "debug.h"
86 #include "chat.h"
87 #include "distributed/distributed.h"
88 #include "distributed/merge.h"
90 /* Internal engine state. */
91 struct distributed {
92 char *slave_port;
93 char *proxy_port;
94 int max_slaves;
95 int shared_nodes;
96 int stats_hbits;
97 bool slaves_quit;
98 struct move my_last_move;
99 struct move_stats my_last_stats;
100 int slaves;
101 int threads;
104 /* Default number of simulations to perform per move.
105 * Note that this is in total over all slaves! */
106 #define DIST_GAMES 80000
107 static const struct time_info default_ti = {
108 .period = TT_MOVE,
109 .dim = TD_GAMES,
110 .len = { .games = DIST_GAMES },
113 #define get_value(value, color) \
114 ((color) == S_BLACK ? (value) : 1 - (value))
117 /* Maximum time (seconds) to wait for answers to fast gtp commands
118 * (all commands except pachi-genmoves and final_status_list). */
119 #define MAX_FAST_CMD_WAIT 0.5
121 /* Maximum time (seconds) to wait for answers to genmoves. */
122 #define MAX_GENMOVES_WAIT 0.1 /* 100 ms */
124 /* Minimum time (seconds) to wait before we stop early. This should
125 * ensure that most slaves have replied at least once. */
126 #define MIN_EARLY_STOP_WAIT 0.3 /* 300 ms */
128 /* Display a path as leaf<parent<grandparent...
129 * Returns the path string in a static buffer; it is NOT safe for
130 * anything but debugging - in particular, it is NOT thread-safe! */
131 char *
132 path2sstr(path_t path, struct board *b)
134 /* Special case for pass and resign. */
135 if (path < 0) return coord2sstr((coord_t)path, b);
137 static char buf[16][64];
138 static int bi = 0;
139 char *b2;
140 b2 = buf[bi++ & 15];
141 *b2 = '\0';
142 char *s = b2;
143 char *end = b2 + 64;
144 coord_t leaf;
145 while ((leaf = leaf_coord(path, b)) != 0) {
146 s += snprintf(s, end - s, "%s<", coord2sstr(leaf, b));
147 path = parent_path(path, b);
149 if (s != b2) s[-1] = '\0';
150 return b2;
153 /* Dispatch a new gtp command to all slaves.
154 * The slave lock must not be held upon entry and is released upon return.
155 * args is empty or ends with '\n' */
156 static enum parse_code
157 distributed_notify(struct engine *e, struct board *b, int id, char *cmd, char *args, char **reply)
159 struct distributed *dist = e->data;
161 /* Commands that should not be sent to slaves.
162 * time_left will be part of next pachi-genmoves,
163 * we reduce latency by not forwarding it here. */
164 if ((!strcasecmp(cmd, "quit") && !dist->slaves_quit)
165 || !strcasecmp(cmd, "pachi-gentbook")
166 || !strcasecmp(cmd, "pachi-dumptbook")
167 || !strcasecmp(cmd, "kgs-chat")
168 || !strcasecmp(cmd, "time_left")
170 /* and commands that will be sent to slaves later */
171 || !strcasecmp(cmd, "genmove")
172 || !strcasecmp(cmd, "kgs-genmove_cleanup")
173 || !strcasecmp(cmd, "final_score")
174 || !strcasecmp(cmd, "final_status_list"))
175 return P_OK;
177 protocol_lock();
179 // Create a new command to be sent by the slave threads.
180 new_cmd(b, cmd, args);
182 /* Wait for replies here. If we don't wait, we run the
183 * risk of getting out of sync with most slaves and
184 * sending command history too frequently. But don't wait
185 * for all slaves otherwise we can lose on time because of
186 * a single slow slave when replaying a whole game. */
187 int min_slaves = active_slaves > 1 ? 3 * active_slaves / 4 : 1;
188 get_replies(time_now() + MAX_FAST_CMD_WAIT, min_slaves);
190 protocol_unlock();
192 // At the beginning wait even more for late slaves.
193 if (b->moves == 0) sleep(1);
194 return P_OK;
197 /* The playouts sent by slaves for the children of the root node
198 * include contributions from other slaves. To avoid 32-bit overflow on
199 * large configurations with many slaves we must average the playouts. */
200 struct large_stats {
201 long playouts; // # of playouts
202 floating_t value; // BLACK wins/playouts
205 static void
206 large_stats_add_result(struct large_stats *s, floating_t result, long playouts)
208 s->playouts += playouts;
209 s->value += (result - s->value) * playouts / s->playouts;
212 /* genmoves returns "=id played_own total_playouts threads keep_looking @size"
213 * then a list of lines "coord playouts value" with absolute counts for
214 * children of the root node, then a binary array of incr_stats structs.
215 * To simplify the code, we assume that master and slave have the same architecture
216 * (store values identically).
217 * Return the move with most playouts, and additional stats.
218 * keep_looking is set from a majority vote of the slaves seen so far for this
219 * move but should not be trusted if too few slaves have been seen.
220 * Keep this code in sync with uct/slave.c:report_stats().
221 * slave_lock is held on entry and on return. */
222 static coord_t
223 select_best_move(struct board *b, struct large_stats *stats, int *played,
224 int *total_playouts, int *total_threads, bool *keep_looking)
226 assert(reply_count > 0);
228 /* +2 for pass and resign */
229 memset(stats-2, 0, (board_size2(b)+2) * sizeof(*stats));
231 coord_t best_move = pass;
232 long best_playouts = 0;
233 *played = 0;
234 *total_playouts = 0;
235 *total_threads = 0;
236 int keep = 0;
238 for (int reply = 0; reply < reply_count; reply++) {
239 char *r = gtp_replies[reply];
240 int id, o, p, t, k;
241 if (sscanf(r, "=%d %d %d %d %d", &id, &o, &p, &t, &k) != 5) continue;
242 *played += o;
243 *total_playouts += p;
244 *total_threads += t;
245 keep += k;
246 // Skip the rest of the firt line in particular @size
247 r = strchr(r, '\n');
249 char move[64];
250 struct move_stats s;
251 while (r && sscanf(++r, "%63s %d " PRIfloating, move, &s.playouts, &s.value) == 3) {
252 coord_t c = str2scoord(move, board_size(b));
253 assert (c >= resign && c < board_size2(b) && s.playouts >= 0);
255 large_stats_add_result(&stats[c], s.value, (long)s.playouts);
257 if (stats[c].playouts > best_playouts) {
258 best_playouts = stats[c].playouts;
259 best_move = c;
261 r = strchr(r, '\n');
264 for (coord_t c = resign; c < board_size2(b); c++)
265 stats[c].playouts /= reply_count;
266 *keep_looking = keep > reply_count / 2;
267 return best_move;
270 /* Set the args for the genmoves command. If binary_args is set,
271 * each slave thred will add the correct binary size when sending
272 * (see get_binary_arg()). args must have CMDS_SIZE bytes and
273 * upon return ends with a single \n.
274 * Keep this code in sync with uct/slave.c:uct_genmoves().
275 * slave_lock is held on entry and on return but we don't
276 * rely on the lock here. */
277 static void
278 genmoves_args(char *args, enum stone color, int played,
279 struct time_info *ti, bool binary_args)
281 char *end = args + CMDS_SIZE;
282 char *s = args + snprintf(args, CMDS_SIZE, "%s %d", stone2str(color), played);
284 if (ti->dim == TD_WALLTIME) {
285 s += snprintf(s, end - s, " %.3f %.3f %d %d",
286 ti->len.t.main_time, ti->len.t.byoyomi_time,
287 ti->len.t.byoyomi_periods, ti->len.t.byoyomi_stones);
289 s += snprintf(s, end - s, binary_args ? " @0\n" : "\n");
292 /* Time control is mostly done by the slaves, so we use default values here. */
293 #define FUSEKI_END 20
294 #define YOSE_START 40
295 #define MAX_MAINTIME_RATIO 3.0
297 /* Regularly send genmoves command to the slaves, and select the best move. */
298 static coord_t *
299 distributed_genmove(struct engine *e, struct board *b, struct time_info *ti,
300 enum stone color, bool pass_all_alive)
302 struct distributed *dist = e->data;
303 double now = time_now();
304 double first = now;
305 char buf[BSIZE]; // debug only
307 char *cmd = pass_all_alive ? "pachi-genmoves_cleanup" : "pachi-genmoves";
308 char args[CMDS_SIZE];
310 coord_t best;
311 int played, playouts, threads;
313 if (ti->period == TT_NULL) *ti = default_ti;
314 struct time_stop stop;
315 time_stop_conditions(ti, b, FUSEKI_END, YOSE_START, MAX_MAINTIME_RATIO, &stop);
316 struct time_info saved_ti = *ti;
318 /* Combined move stats from all slaves, only for children
319 * of the root node, plus 2 for pass and resign. */
320 struct large_stats stats_array[board_size2(b) + 2], *stats;
321 stats = &stats_array[2];
323 protocol_lock();
324 clear_receive_queue();
326 /* Send the first genmoves without stats. */
327 genmoves_args(args, color, 0, ti, false);
328 new_cmd(b, cmd, args);
330 /* Loop until most slaves want to quit or time elapsed. */
331 int iterations;
332 for (iterations = 1; ; iterations++) {
333 double start = now;
334 /* Wait for just one slave to get stats as fresh as possible,
335 * or at most 100ms to check if we run out of time. */
336 get_replies(now + MAX_GENMOVES_WAIT, 1);
337 now = time_now();
338 if (ti->dim == TD_WALLTIME)
339 time_sub(ti, now - start, false);
341 bool keep_looking;
342 best = select_best_move(b, stats, &played, &playouts, &threads, &keep_looking);
344 if (ti->dim == TD_WALLTIME) {
345 if (now - ti->len.t.timer_start >= stop.worst.time) break;
346 if (!keep_looking && now - first >= MIN_EARLY_STOP_WAIT) break;
347 } else {
348 if (!keep_looking || played >= stop.worst.playouts) break;
350 if (DEBUGVV(2)) {
351 char *coord = coord2sstr(best, b);
352 snprintf(buf, sizeof(buf),
353 "temp winner is %s %s with score %1.4f (%d/%d games)"
354 " %d slaves %d threads\n",
355 stone2str(color), coord, get_value(stats[best].value, color),
356 (int)stats[best].playouts, playouts, reply_count, threads);
357 logline(NULL, "* ", buf);
359 /* Send the command with the same gtp id, to avoid discarding
360 * a reply to a previous genmoves at the same move. */
361 genmoves_args(args, color, played, ti, true);
362 update_cmd(b, cmd, args, false);
364 int replies = reply_count;
366 /* Do not subtract time spent twice (see gtp_parse). */
367 *ti = saved_ti;
369 dist->my_last_move.color = color;
370 dist->my_last_move.coord = best;
371 dist->my_last_stats.value = stats[best].value;
372 dist->my_last_stats.playouts = (int)stats[best].playouts;
373 dist->slaves = reply_count;
374 dist->threads = threads;
376 /* Tell the slaves to commit to the selected move, overwriting
377 * the last "pachi-genmoves" in the command history. */
378 clear_receive_queue();
379 char coordbuf[4];
380 char *coord = coord2bstr(coordbuf, best, b);
381 snprintf(args, sizeof(args), "%s %s\n", stone2str(color), coord);
382 update_cmd(b, "play", args, true);
383 protocol_unlock();
385 if (DEBUGL(1)) {
386 double time = now - first + 0.000001; /* avoid divide by zero */
387 snprintf(buf, sizeof(buf),
388 "GLOBAL WINNER is %s %s with score %1.4f (%d/%d games)\n"
389 "genmove %d games in %0.2fs %d slaves %d threads (%d games/s,"
390 " %d games/s/slave, %d games/s/thread, %.3f ms/iter)\n",
391 stone2str(color), coord, get_value(stats[best].value, color),
392 (int)stats[best].playouts, playouts, played, time, replies, threads,
393 (int)(played/time), (int)(played/time/replies),
394 (int)(played/time/threads), 1000*time/iterations);
395 logline(NULL, "* ", buf);
397 if (DEBUGL(3)) {
398 int total_hnodes = replies * (1 << dist->stats_hbits);
399 merge_print_stats(total_hnodes);
401 return coord_copy(best);
404 static char *
405 distributed_chat(struct engine *e, struct board *b, bool opponent, char *from, char *cmd)
407 struct distributed *dist = e->data;
408 double winrate = get_value(dist->my_last_stats.value, dist->my_last_move.color);
410 return generic_chat(b, opponent, from, cmd, dist->my_last_move.color, dist->my_last_move.coord,
411 dist->my_last_stats.playouts, dist->slaves, dist->threads, winrate, 0.0);
414 static int
415 scmp(const void *p1, const void *p2)
417 return strcasecmp(*(char * const *)p1, *(char * const *)p2);
420 static void
421 distributed_dead_group_list(struct engine *e, struct board *b, struct move_queue *mq)
423 protocol_lock();
425 new_cmd(b, "final_status_list", "dead\n");
426 get_replies(time_now() + MAX_FAST_CMD_WAIT, active_slaves);
428 /* Find the most popular reply. */
429 qsort(gtp_replies, reply_count, sizeof(char *), scmp);
430 int best_reply = 0;
431 int best_count = 1;
432 int count = 1;
433 for (int reply = 1; reply < reply_count; reply++) {
434 if (!strcmp(gtp_replies[reply], gtp_replies[reply-1])) {
435 count++;
436 } else {
437 count = 1;
439 if (count > best_count) {
440 best_count = count;
441 best_reply = reply;
445 /* Pick the first move of each line as group. */
446 char *dead = gtp_replies[best_reply];
447 dead = strchr(dead, ' '); // skip "id "
448 while (dead && *++dead != '\n') {
449 mq_add(mq, str2scoord(dead, board_size(b)), 0);
450 dead = strchr(dead, '\n');
452 protocol_unlock();
455 static struct distributed *
456 distributed_state_init(char *arg, struct board *b)
458 struct distributed *dist = calloc2(1, sizeof(struct distributed));
460 dist->stats_hbits = DEFAULT_STATS_HBITS;
461 dist->max_slaves = DEFAULT_MAX_SLAVES;
462 dist->shared_nodes = DEFAULT_SHARED_NODES;
463 if (arg) {
464 char *optspec, *next = arg;
465 while (*next) {
466 optspec = next;
467 next += strcspn(next, ",");
468 if (*next) { *next++ = 0; } else { *next = 0; }
470 char *optname = optspec;
471 char *optval = strchr(optspec, '=');
472 if (optval) *optval++ = 0;
474 if (!strcasecmp(optname, "slave_port") && optval) {
475 dist->slave_port = strdup(optval);
476 } else if (!strcasecmp(optname, "proxy_port") && optval) {
477 dist->proxy_port = strdup(optval);
478 } else if (!strcasecmp(optname, "max_slaves") && optval) {
479 dist->max_slaves = atoi(optval);
480 } else if (!strcasecmp(optname, "shared_nodes") && optval) {
481 /* Share at most shared_nodes between master and slave at each genmoves.
482 * Must use the same value in master and slaves. */
483 dist->shared_nodes = atoi(optval);
484 } else if (!strcasecmp(optname, "stats_hbits") && optval) {
485 /* Set hash table size to 2^stats_hbits for the shared stats. */
486 dist->stats_hbits = atoi(optval);
487 } else if (!strcasecmp(optname, "slaves_quit")) {
488 dist->slaves_quit = !optval || atoi(optval);
489 } else {
490 fprintf(stderr, "distributed: Invalid engine argument %s or missing value\n", optname);
495 gtp_replies = calloc2(dist->max_slaves, sizeof(char *));
497 if (!dist->slave_port) {
498 fprintf(stderr, "distributed: missing slave_port\n");
499 exit(1);
502 merge_init(&default_sstate, dist->shared_nodes, dist->stats_hbits, dist->max_slaves);
503 protocol_init(dist->slave_port, dist->proxy_port, dist->max_slaves);
505 return dist;
508 struct engine *
509 engine_distributed_init(char *arg, struct board *b)
511 struct distributed *dist = distributed_state_init(arg, b);
512 struct engine *e = calloc2(1, sizeof(struct engine));
513 e->name = "Distributed";
514 e->comment = "If you believe you have won but I am still playing, "
515 "please help me understand by capturing all dead stones. "
516 "Anyone can send me 'winrate' in private chat to get my assessment of the position.";
517 e->notify = distributed_notify;
518 e->genmove = distributed_genmove;
519 e->dead_group_list = distributed_dead_group_list;
520 e->chat = distributed_chat;
521 e->data = dist;
522 // Keep the threads and the open socket connections:
523 e->keep_on_clear = true;
525 return e;