Merge branch 'derm6'
[pachi/json.git] / distributed / distributed.c
blob0bc9887456d38e1f59822dcf23f5765b5a2fee1d
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 * zzgo -e distributed slave_port=1234
59 * and N slaves running as:
60 * zzgo -e uct -g masterhost:1234 slave
61 * With log proxy:
62 * zzgo -e distributed slave_port=1234,proxy_port=1235
63 * zzgo -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 * zzgo -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 <alloca.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 "distributed/distributed.h"
87 #include "distributed/merge.h"
89 /* Internal engine state. */
90 struct distributed {
91 char *slave_port;
92 char *proxy_port;
93 int max_slaves;
94 int shared_nodes;
95 int stats_hbits;
96 bool slaves_quit;
97 struct move my_last_move;
98 struct move_stats my_last_stats;
101 /* Default number of simulations to perform per move.
102 * Note that this is in total over all slaves! */
103 #define DIST_GAMES 80000
104 static const struct time_info default_ti = {
105 .period = TT_MOVE,
106 .dim = TD_GAMES,
107 .len = { .games = DIST_GAMES },
110 #define get_value(value, color) \
111 ((color) == S_BLACK ? (value) : 1 - (value))
114 /* Maximum time (seconds) to wait for answers to fast gtp commands
115 * (all commands except pachi-genmoves and final_status_list). */
116 #define MAX_FAST_CMD_WAIT 1.0
118 /* Maximum time (seconds) to wait for answers to genmoves. */
119 #define MAX_GENMOVES_WAIT 0.1 /* 100 ms */
121 /* Minimum time (seconds) to wait before we stop early. This should
122 * ensure that most slaves have replied at least once. */
123 #define MIN_EARLY_STOP_WAIT 0.3 /* 300 ms */
125 /* Display a path as leaf<parent<grandparent...
126 * Returns the path string in a static buffer; it is NOT safe for
127 * anything but debugging - in particular, it is NOT thread-safe! */
128 char *
129 path2sstr(path_t path, struct board *b)
131 /* Special case for pass and resign. */
132 if (path < 0) return coord2sstr((coord_t)path, b);
134 static char buf[16][64];
135 static int bi = 0;
136 char *b2;
137 b2 = buf[bi++ & 15];
138 *b2 = '\0';
139 char *s = b2;
140 char *end = b2 + 64;
141 coord_t leaf;
142 while ((leaf = leaf_coord(path, b)) != 0) {
143 s += snprintf(s, end - s, "%s<", coord2sstr(leaf, b));
144 path = parent_path(path, b);
146 if (s != b2) s[-1] = '\0';
147 return b2;
150 /* Dispatch a new gtp command to all slaves.
151 * The slave lock must not be held upon entry and is released upon return.
152 * args is empty or ends with '\n' */
153 static enum parse_code
154 distributed_notify(struct engine *e, struct board *b, int id, char *cmd, char *args, char **reply)
156 struct distributed *dist = e->data;
158 /* Commands that should not be sent to slaves.
159 * time_left will be part of next pachi-genmoves,
160 * we reduce latency by not forwarding it here. */
161 if ((!strcasecmp(cmd, "quit") && !dist->slaves_quit)
162 || !strcasecmp(cmd, "uct_gentbook")
163 || !strcasecmp(cmd, "uct_dumptbook")
164 || !strcasecmp(cmd, "kgs-chat")
165 || !strcasecmp(cmd, "time_left")
167 /* and commands that will be sent to slaves later */
168 || !strcasecmp(cmd, "genmove")
169 || !strcasecmp(cmd, "kgs-genmove_cleanup")
170 || !strcasecmp(cmd, "final_score")
171 || !strcasecmp(cmd, "final_status_list"))
172 return P_OK;
174 protocol_lock();
176 // Create a new command to be sent by the slave threads.
177 new_cmd(b, cmd, args);
179 /* Wait for replies here. If we don't wait, we run the
180 * risk of getting out of sync with most slaves and
181 * sending command history too frequently. */
182 get_replies(time_now() + MAX_FAST_CMD_WAIT, active_slaves);
184 protocol_unlock();
185 return P_OK;
188 /* genmoves returns "=id played_own total_playouts threads keep_looking @size"
189 * then a list of lines "coord playouts value" with absolute counts for
190 * children of the root node, then a binary array of incr_stats structs.
191 * To simplify the code, we assume that master and slave have the same architecture
192 * (store values identically).
193 * Return the move with most playouts, and additional stats.
194 * keep_looking is set from a majority vote of the slaves seen so far for this
195 * move but should not be trusted if too few slaves have been seen.
196 * Keep this code in sync with uct/slave.c:report_stats().
197 * slave_lock is held on entry and on return. */
198 static coord_t
199 select_best_move(struct board *b, struct move_stats *stats, int *played,
200 int *total_playouts, int *total_threads, bool *keep_looking)
202 assert(reply_count > 0);
204 /* +2 for pass and resign */
205 memset(stats-2, 0, (board_size2(b)+2) * sizeof(*stats));
207 coord_t best_move = pass;
208 int best_playouts = -1;
209 *played = 0;
210 *total_playouts = 0;
211 *total_threads = 0;
212 int keep = 0;
214 for (int reply = 0; reply < reply_count; reply++) {
215 char *r = gtp_replies[reply];
216 int id, o, p, t, k;
217 if (sscanf(r, "=%d %d %d %d %d", &id, &o, &p, &t, &k) != 5) continue;
218 *played += o;
219 *total_playouts += p;
220 *total_threads += t;
221 keep += k;
222 // Skip the rest of the firt line in particular @size
223 r = strchr(r, '\n');
225 char move[64];
226 struct move_stats s;
227 while (r && sscanf(++r, "%63s %d %f", move, &s.playouts, &s.value) == 3) {
228 coord_t c = str2scoord(move, board_size(b));
229 assert (c >= resign && c < board_size2(b) && s.playouts >= 0);
231 stats_add_result(&stats[c], s.value, s.playouts);
233 if (stats[c].playouts > best_playouts) {
234 best_playouts = stats[c].playouts;
235 best_move = c;
237 r = strchr(r, '\n');
240 *keep_looking = keep > reply_count / 2;
241 return best_move;
244 /* Set the args for the genmoves command. If binary_args is set,
245 * each slave thred will add the correct binary size when sending
246 * (see get_binary_arg()). args must have CMDS_SIZE bytes and
247 * upon return ends with a single \n.
248 * Keep this code in sync with uct/slave.c:uct_genmoves().
249 * slave_lock is held on entry and on return but we don't
250 * rely on the lock here. */
251 static void
252 genmoves_args(char *args, enum stone color, int played,
253 struct time_info *ti, bool binary_args)
255 char *end = args + CMDS_SIZE;
256 char *s = args + snprintf(args, CMDS_SIZE, "%s %d", stone2str(color), played);
258 if (ti->dim == TD_WALLTIME) {
259 s += snprintf(s, end - s, " %.3f %.3f %d %d",
260 ti->len.t.main_time, ti->len.t.byoyomi_time,
261 ti->len.t.byoyomi_periods, ti->len.t.byoyomi_stones);
263 s += snprintf(s, end - s, binary_args ? " @0\n" : "\n");
266 /* Time control is mostly done by the slaves, so we use default values here. */
267 #define FUSEKI_END 20
268 #define YOSE_START 40
270 /* In the ascii reply to genmoves, each slave sends absolute counts
271 * including contributions from other slaves. For human display
272 * reduce the sum to an average. */
273 #define stats_average(playouts) ((playouts) / reply_count)
275 /* Regularly send genmoves command to the slaves, and select the best move. */
276 static coord_t *
277 distributed_genmove(struct engine *e, struct board *b, struct time_info *ti,
278 enum stone color, bool pass_all_alive)
280 struct distributed *dist = e->data;
281 double now = time_now();
282 double first = now;
283 char buf[BSIZE]; // debug only
285 char *cmd = pass_all_alive ? "pachi-genmoves_cleanup" : "pachi-genmoves";
286 char args[CMDS_SIZE];
288 coord_t best;
289 int played, playouts, threads;
291 if (ti->period == TT_NULL) *ti = default_ti;
292 struct time_stop stop;
293 time_stop_conditions(ti, b, FUSEKI_END, YOSE_START, &stop);
294 struct time_info saved_ti = *ti;
296 /* Combined move stats from all slaves, only for children
297 * of the root node, plus 2 for pass and resign. */
298 struct move_stats *stats = alloca((board_size2(b)+2) * sizeof(struct move_stats));
299 stats += 2;
301 protocol_lock();
302 clear_receive_queue();
304 /* Send the first genmoves without stats. */
305 genmoves_args(args, color, 0, ti, false);
306 new_cmd(b, cmd, args);
308 /* Loop until most slaves want to quit or time elapsed. */
309 int iterations;
310 for (iterations = 1; ; iterations++) {
311 double start = now;
312 /* Wait for just one slave to get stats as fresh as possible,
313 * or at most 100ms to check if we run out of time. */
314 get_replies(now + MAX_GENMOVES_WAIT, 1);
315 now = time_now();
316 if (ti->dim == TD_WALLTIME)
317 time_sub(ti, now - start, false);
319 bool keep_looking;
320 best = select_best_move(b, stats, &played, &playouts, &threads, &keep_looking);
322 if (ti->dim == TD_WALLTIME) {
323 if (now - ti->len.t.timer_start >= stop.worst.time) break;
324 if (!keep_looking && now - first >= MIN_EARLY_STOP_WAIT) break;
325 } else {
326 if (!keep_looking || played >= stop.worst.playouts) break;
328 if (DEBUGVV(2)) {
329 char *coord = coord2sstr(best, b);
330 snprintf(buf, sizeof(buf),
331 "temp winner is %s %s with score %1.4f (%d/%d games)"
332 " %d slaves %d threads\n",
333 stone2str(color), coord, get_value(stats[best].value, color),
334 stats_average(stats[best].playouts), playouts, reply_count, threads);
335 logline(NULL, "* ", buf);
337 /* Send the command with the same gtp id, to avoid discarding
338 * a reply to a previous genmoves at the same move. */
339 genmoves_args(args, color, played, ti, true);
340 update_cmd(b, cmd, args, false);
342 int replies = reply_count;
344 /* Do not subtract time spent twice (see gtp_parse). */
345 *ti = saved_ti;
347 dist->my_last_move.color = color;
348 dist->my_last_move.coord = best;
349 dist->my_last_stats.value = stats[best].value;
350 dist->my_last_stats.playouts = stats_average(stats[best].playouts);
352 /* Tell the slaves to commit to the selected move, overwriting
353 * the last "pachi-genmoves" in the command history. */
354 clear_receive_queue();
355 char coordbuf[4];
356 char *coord = coord2bstr(coordbuf, best, b);
357 snprintf(args, sizeof(args), "%s %s\n", stone2str(color), coord);
358 update_cmd(b, "play", args, true);
359 protocol_unlock();
361 if (DEBUGL(1)) {
362 double time = now - first + 0.000001; /* avoid divide by zero */
363 snprintf(buf, sizeof(buf),
364 "GLOBAL WINNER is %s %s with score %1.4f (%d/%d games)\n"
365 "genmove %d games in %0.2fs %d slaves %d threads (%d games/s,"
366 " %d games/s/slave, %d games/s/thread, %.3f ms/iter)\n",
367 stone2str(color), coord, get_value(stats[best].value, color),
368 stats[best].playouts, playouts, played, time, replies, threads,
369 (int)(played/time), (int)(played/time/replies),
370 (int)(played/time/threads), 1000*time/iterations);
371 logline(NULL, "* ", buf);
373 if (DEBUGL(3)) {
374 int total_hnodes = replies * (1 << dist->stats_hbits);
375 merge_print_stats(total_hnodes);
377 return coord_copy(best);
380 static char *
381 distributed_chat(struct engine *e, struct board *b, char *cmd)
383 struct distributed *dist = e->data;
384 static char reply[BSIZE];
386 cmd += strspn(cmd, " \n\t");
387 if (!strncasecmp(cmd, "winrate", 7)) {
388 enum stone color = dist->my_last_move.color;
389 snprintf(reply, BSIZE, "In %d playouts at %d machines, %s %s can win with %.2f%% probability.",
390 dist->my_last_stats.playouts, active_slaves, stone2str(color),
391 coord2sstr(dist->my_last_move.coord, b),
392 100 * get_value(dist->my_last_stats.value, color));
393 return reply;
395 return NULL;
398 static int
399 scmp(const void *p1, const void *p2)
401 return strcasecmp(*(char * const *)p1, *(char * const *)p2);
404 static void
405 distributed_dead_group_list(struct engine *e, struct board *b, struct move_queue *mq)
407 protocol_lock();
409 new_cmd(b, "final_status_list", "dead\n");
410 get_replies(time_now() + MAX_FAST_CMD_WAIT, active_slaves);
412 /* Find the most popular reply. */
413 qsort(gtp_replies, reply_count, sizeof(char *), scmp);
414 int best_reply = 0;
415 int best_count = 1;
416 int count = 1;
417 for (int reply = 1; reply < reply_count; reply++) {
418 if (!strcmp(gtp_replies[reply], gtp_replies[reply-1])) {
419 count++;
420 } else {
421 count = 1;
423 if (count > best_count) {
424 best_count = count;
425 best_reply = reply;
429 /* Pick the first move of each line as group. */
430 char *dead = gtp_replies[best_reply];
431 dead = strchr(dead, ' '); // skip "id "
432 while (dead && *++dead != '\n') {
433 mq_add(mq, str2scoord(dead, board_size(b)), 0);
434 dead = strchr(dead, '\n');
436 protocol_unlock();
439 static struct distributed *
440 distributed_state_init(char *arg, struct board *b)
442 struct distributed *dist = calloc2(1, sizeof(struct distributed));
444 dist->stats_hbits = DEFAULT_STATS_HBITS;
445 dist->max_slaves = DEFAULT_MAX_SLAVES;
446 dist->shared_nodes = DEFAULT_SHARED_NODES;
447 if (arg) {
448 char *optspec, *next = arg;
449 while (*next) {
450 optspec = next;
451 next += strcspn(next, ",");
452 if (*next) { *next++ = 0; } else { *next = 0; }
454 char *optname = optspec;
455 char *optval = strchr(optspec, '=');
456 if (optval) *optval++ = 0;
458 if (!strcasecmp(optname, "slave_port") && optval) {
459 dist->slave_port = strdup(optval);
460 } else if (!strcasecmp(optname, "proxy_port") && optval) {
461 dist->proxy_port = strdup(optval);
462 } else if (!strcasecmp(optname, "max_slaves") && optval) {
463 dist->max_slaves = atoi(optval);
464 } else if (!strcasecmp(optname, "shared_nodes") && optval) {
465 /* Share at most shared_nodes between master and slave at each genmoves.
466 * Must use the same value in master and slaves. */
467 dist->shared_nodes = atoi(optval);
468 } else if (!strcasecmp(optname, "stats_hbits") && optval) {
469 /* Set hash table size to 2^stats_hbits for the shared stats. */
470 dist->stats_hbits = atoi(optval);
471 } else if (!strcasecmp(optname, "slaves_quit")) {
472 dist->slaves_quit = !optval || atoi(optval);
473 } else {
474 fprintf(stderr, "distributed: Invalid engine argument %s or missing value\n", optname);
479 gtp_replies = calloc2(dist->max_slaves, sizeof(char *));
481 if (!dist->slave_port) {
482 fprintf(stderr, "distributed: missing slave_port\n");
483 exit(1);
486 merge_init(&default_sstate, dist->shared_nodes, dist->stats_hbits, dist->max_slaves);
487 protocol_init(dist->slave_port, dist->proxy_port, dist->max_slaves);
489 return dist;
492 struct engine *
493 engine_distributed_init(char *arg, struct board *b)
495 struct distributed *dist = distributed_state_init(arg, b);
496 struct engine *e = calloc2(1, sizeof(struct engine));
497 e->name = "Distributed Engine";
498 e->comment = "I'm playing the distributed engine. When I'm losing, I will resign, "
499 "if I think I win, I play until you pass. "
500 "Anyone can send me 'winrate' in private chat to get my assessment of the position.";
501 e->notify = distributed_notify;
502 e->genmove = distributed_genmove;
503 e->dead_group_list = distributed_dead_group_list;
504 e->chat = distributed_chat;
505 e->data = dist;
506 // Keep the threads and the open socket connections:
507 e->keep_on_clear = true;
509 return e;