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
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
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
76 #include <sys/types.h>
87 #include "distributed/distributed.h"
88 #include "distributed/merge.h"
90 /* Internal engine state. */
98 struct move my_last_move
;
99 struct move_stats my_last_stats
;
102 /* Default number of simulations to perform per move.
103 * Note that this is in total over all slaves! */
104 #define DIST_GAMES 80000
105 static const struct time_info default_ti
= {
108 .len
= { .games
= DIST_GAMES
},
111 #define get_value(value, color) \
112 ((color) == S_BLACK ? (value) : 1 - (value))
115 /* Maximum time (seconds) to wait for answers to fast gtp commands
116 * (all commands except pachi-genmoves and final_status_list). */
117 #define MAX_FAST_CMD_WAIT 1.0
119 /* Maximum time (seconds) to wait for answers to genmoves. */
120 #define MAX_GENMOVES_WAIT 0.1 /* 100 ms */
122 /* Minimum time (seconds) to wait before we stop early. This should
123 * ensure that most slaves have replied at least once. */
124 #define MIN_EARLY_STOP_WAIT 0.3 /* 300 ms */
126 /* Display a path as leaf<parent<grandparent...
127 * Returns the path string in a static buffer; it is NOT safe for
128 * anything but debugging - in particular, it is NOT thread-safe! */
130 path2sstr(path_t path
, struct board
*b
)
132 /* Special case for pass and resign. */
133 if (path
< 0) return coord2sstr((coord_t
)path
, b
);
135 static char buf
[16][64];
143 while ((leaf
= leaf_coord(path
, b
)) != 0) {
144 s
+= snprintf(s
, end
- s
, "%s<", coord2sstr(leaf
, b
));
145 path
= parent_path(path
, b
);
147 if (s
!= b2
) s
[-1] = '\0';
151 /* Dispatch a new gtp command to all slaves.
152 * The slave lock must not be held upon entry and is released upon return.
153 * args is empty or ends with '\n' */
154 static enum parse_code
155 distributed_notify(struct engine
*e
, struct board
*b
, int id
, char *cmd
, char *args
, char **reply
)
157 struct distributed
*dist
= e
->data
;
159 /* Commands that should not be sent to slaves.
160 * time_left will be part of next pachi-genmoves,
161 * we reduce latency by not forwarding it here. */
162 if ((!strcasecmp(cmd
, "quit") && !dist
->slaves_quit
)
163 || !strcasecmp(cmd
, "uct_gentbook")
164 || !strcasecmp(cmd
, "uct_dumptbook")
165 || !strcasecmp(cmd
, "kgs-chat")
166 || !strcasecmp(cmd
, "time_left")
168 /* and commands that will be sent to slaves later */
169 || !strcasecmp(cmd
, "genmove")
170 || !strcasecmp(cmd
, "kgs-genmove_cleanup")
171 || !strcasecmp(cmd
, "final_score")
172 || !strcasecmp(cmd
, "final_status_list"))
177 // Create a new command to be sent by the slave threads.
178 new_cmd(b
, cmd
, args
);
180 /* Wait for replies here. If we don't wait, we run the
181 * risk of getting out of sync with most slaves and
182 * sending command history too frequently. */
183 get_replies(time_now() + MAX_FAST_CMD_WAIT
, active_slaves
);
187 // At the beginning wait even more for late slaves.
188 if (b
->moves
== 0) sleep(1);
192 /* The playouts sent by slaves for the children of the root node
193 * include contributions from other slaves. To avoid 32-bit overflow on
194 * large configurations with many slaves we must average the playouts. */
196 long playouts
; // # of playouts
197 floating_t value
; // BLACK wins/playouts
201 large_stats_add_result(struct large_stats
*s
, floating_t result
, long playouts
)
203 s
->playouts
+= playouts
;
204 s
->value
+= (result
- s
->value
) * playouts
/ s
->playouts
;
207 /* genmoves returns "=id played_own total_playouts threads keep_looking @size"
208 * then a list of lines "coord playouts value" with absolute counts for
209 * children of the root node, then a binary array of incr_stats structs.
210 * To simplify the code, we assume that master and slave have the same architecture
211 * (store values identically).
212 * Return the move with most playouts, and additional stats.
213 * keep_looking is set from a majority vote of the slaves seen so far for this
214 * move but should not be trusted if too few slaves have been seen.
215 * Keep this code in sync with uct/slave.c:report_stats().
216 * slave_lock is held on entry and on return. */
218 select_best_move(struct board
*b
, struct large_stats
*stats
, int *played
,
219 int *total_playouts
, int *total_threads
, bool *keep_looking
)
221 assert(reply_count
> 0);
223 /* +2 for pass and resign */
224 memset(stats
-2, 0, (board_size2(b
)+2) * sizeof(*stats
));
226 coord_t best_move
= pass
;
227 long best_playouts
= -1;
233 for (int reply
= 0; reply
< reply_count
; reply
++) {
234 char *r
= gtp_replies
[reply
];
236 if (sscanf(r
, "=%d %d %d %d %d", &id
, &o
, &p
, &t
, &k
) != 5) continue;
238 *total_playouts
+= p
;
241 // Skip the rest of the firt line in particular @size
246 while (r
&& sscanf(++r
, "%63s %d " PRIfloating
, move
, &s
.playouts
, &s
.value
) == 3) {
247 coord_t c
= str2scoord(move
, board_size(b
));
248 assert (c
>= resign
&& c
< board_size2(b
) && s
.playouts
>= 0);
250 large_stats_add_result(&stats
[c
], s
.value
, (long)s
.playouts
);
252 if (stats
[c
].playouts
> best_playouts
) {
253 best_playouts
= stats
[c
].playouts
;
259 for (coord_t c
= resign
; c
< board_size2(b
); c
++)
260 stats
[c
].playouts
/= reply_count
;
261 *keep_looking
= keep
> reply_count
/ 2;
265 /* Set the args for the genmoves command. If binary_args is set,
266 * each slave thred will add the correct binary size when sending
267 * (see get_binary_arg()). args must have CMDS_SIZE bytes and
268 * upon return ends with a single \n.
269 * Keep this code in sync with uct/slave.c:uct_genmoves().
270 * slave_lock is held on entry and on return but we don't
271 * rely on the lock here. */
273 genmoves_args(char *args
, enum stone color
, int played
,
274 struct time_info
*ti
, bool binary_args
)
276 char *end
= args
+ CMDS_SIZE
;
277 char *s
= args
+ snprintf(args
, CMDS_SIZE
, "%s %d", stone2str(color
), played
);
279 if (ti
->dim
== TD_WALLTIME
) {
280 s
+= snprintf(s
, end
- s
, " %.3f %.3f %d %d",
281 ti
->len
.t
.main_time
, ti
->len
.t
.byoyomi_time
,
282 ti
->len
.t
.byoyomi_periods
, ti
->len
.t
.byoyomi_stones
);
284 s
+= snprintf(s
, end
- s
, binary_args
? " @0\n" : "\n");
287 /* Time control is mostly done by the slaves, so we use default values here. */
288 #define FUSEKI_END 20
289 #define YOSE_START 40
290 #define MAX_MAINTIME_RATIO 3.0
292 /* Regularly send genmoves command to the slaves, and select the best move. */
294 distributed_genmove(struct engine
*e
, struct board
*b
, struct time_info
*ti
,
295 enum stone color
, bool pass_all_alive
)
297 struct distributed
*dist
= e
->data
;
298 double now
= time_now();
300 char buf
[BSIZE
]; // debug only
302 char *cmd
= pass_all_alive
? "pachi-genmoves_cleanup" : "pachi-genmoves";
303 char args
[CMDS_SIZE
];
306 int played
, playouts
, threads
;
308 if (ti
->period
== TT_NULL
) *ti
= default_ti
;
309 struct time_stop stop
;
310 time_stop_conditions(ti
, b
, FUSEKI_END
, YOSE_START
, MAX_MAINTIME_RATIO
, &stop
);
311 struct time_info saved_ti
= *ti
;
313 /* Combined move stats from all slaves, only for children
314 * of the root node, plus 2 for pass and resign. */
315 struct large_stats
*stats
= alloca((board_size2(b
)+2) * sizeof(struct large_stats
));
319 clear_receive_queue();
321 /* Send the first genmoves without stats. */
322 genmoves_args(args
, color
, 0, ti
, false);
323 new_cmd(b
, cmd
, args
);
325 /* Loop until most slaves want to quit or time elapsed. */
327 for (iterations
= 1; ; iterations
++) {
329 /* Wait for just one slave to get stats as fresh as possible,
330 * or at most 100ms to check if we run out of time. */
331 get_replies(now
+ MAX_GENMOVES_WAIT
, 1);
333 if (ti
->dim
== TD_WALLTIME
)
334 time_sub(ti
, now
- start
, false);
337 best
= select_best_move(b
, stats
, &played
, &playouts
, &threads
, &keep_looking
);
339 if (ti
->dim
== TD_WALLTIME
) {
340 if (now
- ti
->len
.t
.timer_start
>= stop
.worst
.time
) break;
341 if (!keep_looking
&& now
- first
>= MIN_EARLY_STOP_WAIT
) break;
343 if (!keep_looking
|| played
>= stop
.worst
.playouts
) break;
346 char *coord
= coord2sstr(best
, b
);
347 snprintf(buf
, sizeof(buf
),
348 "temp winner is %s %s with score %1.4f (%d/%d games)"
349 " %d slaves %d threads\n",
350 stone2str(color
), coord
, get_value(stats
[best
].value
, color
),
351 (int)stats
[best
].playouts
, playouts
, reply_count
, threads
);
352 logline(NULL
, "* ", buf
);
354 /* Send the command with the same gtp id, to avoid discarding
355 * a reply to a previous genmoves at the same move. */
356 genmoves_args(args
, color
, played
, ti
, true);
357 update_cmd(b
, cmd
, args
, false);
359 int replies
= reply_count
;
361 /* Do not subtract time spent twice (see gtp_parse). */
364 dist
->my_last_move
.color
= color
;
365 dist
->my_last_move
.coord
= best
;
366 dist
->my_last_stats
.value
= stats
[best
].value
;
367 dist
->my_last_stats
.playouts
= (int)stats
[best
].playouts
;
369 /* Tell the slaves to commit to the selected move, overwriting
370 * the last "pachi-genmoves" in the command history. */
371 clear_receive_queue();
373 char *coord
= coord2bstr(coordbuf
, best
, b
);
374 snprintf(args
, sizeof(args
), "%s %s\n", stone2str(color
), coord
);
375 update_cmd(b
, "play", args
, true);
379 double time
= now
- first
+ 0.000001; /* avoid divide by zero */
380 snprintf(buf
, sizeof(buf
),
381 "GLOBAL WINNER is %s %s with score %1.4f (%d/%d games)\n"
382 "genmove %d games in %0.2fs %d slaves %d threads (%d games/s,"
383 " %d games/s/slave, %d games/s/thread, %.3f ms/iter)\n",
384 stone2str(color
), coord
, get_value(stats
[best
].value
, color
),
385 (int)stats
[best
].playouts
, playouts
, played
, time
, replies
, threads
,
386 (int)(played
/time
), (int)(played
/time
/replies
),
387 (int)(played
/time
/threads
), 1000*time
/iterations
);
388 logline(NULL
, "* ", buf
);
391 int total_hnodes
= replies
* (1 << dist
->stats_hbits
);
392 merge_print_stats(total_hnodes
);
394 return coord_copy(best
);
398 distributed_chat(struct engine
*e
, struct board
*b
, char *cmd
)
400 struct distributed
*dist
= e
->data
;
401 static char reply
[BSIZE
];
403 cmd
+= strspn(cmd
, " \n\t");
404 if (!strncasecmp(cmd
, "winrate", 7)) {
405 enum stone color
= dist
->my_last_move
.color
;
406 snprintf(reply
, BSIZE
, "In %d playouts at %d machines, %s %s can win with %.2f%% probability.",
407 dist
->my_last_stats
.playouts
, active_slaves
, stone2str(color
),
408 coord2sstr(dist
->my_last_move
.coord
, b
),
409 100 * get_value(dist
->my_last_stats
.value
, color
));
416 scmp(const void *p1
, const void *p2
)
418 return strcasecmp(*(char * const *)p1
, *(char * const *)p2
);
422 distributed_dead_group_list(struct engine
*e
, struct board
*b
, struct move_queue
*mq
)
426 new_cmd(b
, "final_status_list", "dead\n");
427 get_replies(time_now() + MAX_FAST_CMD_WAIT
, active_slaves
);
429 /* Find the most popular reply. */
430 qsort(gtp_replies
, reply_count
, sizeof(char *), scmp
);
434 for (int reply
= 1; reply
< reply_count
; reply
++) {
435 if (!strcmp(gtp_replies
[reply
], gtp_replies
[reply
-1])) {
440 if (count
> best_count
) {
446 /* Pick the first move of each line as group. */
447 char *dead
= gtp_replies
[best_reply
];
448 dead
= strchr(dead
, ' '); // skip "id "
449 while (dead
&& *++dead
!= '\n') {
450 mq_add(mq
, str2scoord(dead
, board_size(b
)), 0);
451 dead
= strchr(dead
, '\n');
456 static struct distributed
*
457 distributed_state_init(char *arg
, struct board
*b
)
459 struct distributed
*dist
= calloc2(1, sizeof(struct distributed
));
461 dist
->stats_hbits
= DEFAULT_STATS_HBITS
;
462 dist
->max_slaves
= DEFAULT_MAX_SLAVES
;
463 dist
->shared_nodes
= DEFAULT_SHARED_NODES
;
465 char *optspec
, *next
= arg
;
468 next
+= strcspn(next
, ",");
469 if (*next
) { *next
++ = 0; } else { *next
= 0; }
471 char *optname
= optspec
;
472 char *optval
= strchr(optspec
, '=');
473 if (optval
) *optval
++ = 0;
475 if (!strcasecmp(optname
, "slave_port") && optval
) {
476 dist
->slave_port
= strdup(optval
);
477 } else if (!strcasecmp(optname
, "proxy_port") && optval
) {
478 dist
->proxy_port
= strdup(optval
);
479 } else if (!strcasecmp(optname
, "max_slaves") && optval
) {
480 dist
->max_slaves
= atoi(optval
);
481 } else if (!strcasecmp(optname
, "shared_nodes") && optval
) {
482 /* Share at most shared_nodes between master and slave at each genmoves.
483 * Must use the same value in master and slaves. */
484 dist
->shared_nodes
= atoi(optval
);
485 } else if (!strcasecmp(optname
, "stats_hbits") && optval
) {
486 /* Set hash table size to 2^stats_hbits for the shared stats. */
487 dist
->stats_hbits
= atoi(optval
);
488 } else if (!strcasecmp(optname
, "slaves_quit")) {
489 dist
->slaves_quit
= !optval
|| atoi(optval
);
491 fprintf(stderr
, "distributed: Invalid engine argument %s or missing value\n", optname
);
496 gtp_replies
= calloc2(dist
->max_slaves
, sizeof(char *));
498 if (!dist
->slave_port
) {
499 fprintf(stderr
, "distributed: missing slave_port\n");
503 merge_init(&default_sstate
, dist
->shared_nodes
, dist
->stats_hbits
, dist
->max_slaves
);
504 protocol_init(dist
->slave_port
, dist
->proxy_port
, dist
->max_slaves
);
510 engine_distributed_init(char *arg
, struct board
*b
)
512 struct distributed
*dist
= distributed_state_init(arg
, b
);
513 struct engine
*e
= calloc2(1, sizeof(struct engine
));
514 e
->name
= "Distributed Engine";
515 e
->comment
= "I'm playing the distributed engine. When I'm losing, I will resign, "
516 "if I think I win, I play until you pass. "
517 "Anyone can send me 'winrate' in private chat to get my assessment of the position.";
518 e
->notify
= distributed_notify
;
519 e
->genmove
= distributed_genmove
;
520 e
->dead_group_list
= distributed_dead_group_list
;
521 e
->chat
= distributed_chat
;
523 // Keep the threads and the open socket connections:
524 e
->keep_on_clear
= true;