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 candidate moves, their number of playouts
7 * and their value. The master then picks the most popular move. */
9 /* With time control, the master waits for all slaves, except
10 * when the allowed time is already passed. In this case the
11 * master picks among the available replies, or waits for just
12 * one reply if there is none yet.
13 * Without time control, the master waits until the desired
14 * number of games have been simulated. In this case the -t
15 * parameter for the master should be the sum of the parameters
18 /* The master sends updated statistics for the best moves
19 * in each genmoves command. In this version only the
20 * children of the root node are updated. The slaves
21 * reply with just their own stats; they remember what was
22 * previously received from or sent to the master, to
23 * distinguish their own contribution from that of other slaves. */
25 /* The master-slave protocol has has fault tolerance. If a slave is
26 * out of sync, the master sends it the appropriate command history. */
28 /* Pass me arguments like a=b,c=d,...
29 * Supported arguments:
30 * slave_port=SLAVE_PORT slaves connect to this port; this parameter is mandatory.
31 * max_slaves=MAX_SLAVES default 100
32 * slaves_quit=0|1 quit gtp command also sent to slaves, default false.
33 * proxy_port=PROXY_PORT slaves optionally send their logs to this port.
34 * Warning: with proxy_port, the master stderr mixes the logs of all
35 * machines but you can separate them again:
36 * slave logs: sed -n '/< .*:/s/.*< /< /p' logfile
37 * master logs: perl -0777 -pe 's/<[ <].*:.*\n//g' logfile
40 /* A configuration without proxy would have one master run on masterhost as:
41 * zzgo -e distributed slave_port=1234
42 * and N slaves running as:
43 * zzgo -e uct -g masterhost:1234 slave
45 * zzgo -e distributed slave_port=1234,proxy_port=1235
46 * zzgo -e uct -g masterhost:1234 -l masterhost:1235 slave
47 * If the master itself runs on a machine other than that running gogui,
48 * gogui-twogtp, kgsGtp or cgosGtp, it can redirect its gtp port:
49 * zzgo -e distributed -g 10000 slave_port=1234,proxy_port=1235
59 #include <sys/types.h>
70 #include "distributed/distributed.h"
71 #include "distributed/protocol.h"
73 /* Internal engine state. */
79 struct move my_last_move
;
80 struct move_stats my_last_stats
;
83 /* Default number of simulations to perform per move.
84 * Note that this is in total over all slaves! */
85 #define DIST_GAMES 80000
86 static const struct time_info default_ti
= {
89 .len
= { .games
= DIST_GAMES
},
92 #define get_value(value, color) \
93 ((color) == S_BLACK ? (value) : 1 - (value))
96 /* Maximum time (seconds) to wait for answers to fast gtp commands
97 * (all commands except pachi-genmoves and final_status_list). */
98 #define MAX_FAST_CMD_WAIT 1.0
100 /* Maximum time (seconds) to wait for answers to genmoves. */
101 #define MAX_GENMOVES_WAIT 0.1 /* 100 ms */
103 /* Dispatch a new gtp command to all slaves.
104 * The slave lock must not be held upon entry and is released upon return.
105 * args is empty or ends with '\n' */
106 static enum parse_code
107 distributed_notify(struct engine
*e
, struct board
*b
, int id
, char *cmd
, char *args
, char **reply
)
109 struct distributed
*dist
= e
->data
;
111 /* Commands that should not be sent to slaves.
112 * time_left will be part of next pachi-genmoves,
113 * we reduce latency by not forwarding it here. */
114 if ((!strcasecmp(cmd
, "quit") && !dist
->slaves_quit
)
115 || !strcasecmp(cmd
, "uct_genbook")
116 || !strcasecmp(cmd
, "uct_dumpbook")
117 || !strcasecmp(cmd
, "kgs-chat")
118 || !strcasecmp(cmd
, "time_left")
120 /* and commands that will be sent to slaves later */
121 || !strcasecmp(cmd
, "genmove")
122 || !strcasecmp(cmd
, "kgs-genmove_cleanup")
123 || !strcasecmp(cmd
, "final_score")
124 || !strcasecmp(cmd
, "final_status_list"))
129 // Create a new command to be sent by the slave threads.
130 new_cmd(b
, cmd
, args
);
132 /* Wait for replies here. If we don't wait, we run the
133 * risk of getting out of sync with most slaves and
134 * sending command history too frequently. */
135 get_replies(time_now() + MAX_FAST_CMD_WAIT
, active_slaves
);
141 /* genmoves returns a line "=id played_own total_playouts threads keep_looking[ reserved]"
142 * then a list of lines "coord playouts value amaf_playouts amaf_value".
143 * Return the move with most playouts, and additional stats.
144 * Keep this code in sync with uct/slave.c:report_stats().
145 * slave_lock is held on entry and on return. */
147 select_best_move(struct board
*b
, struct move_stats2
*stats
, int *played
,
148 int *total_playouts
, int *total_threads
, bool *keep_looking
)
150 assert(reply_count
> 0);
152 /* +2 for pass and resign */
153 memset(stats
-2, 0, (board_size2(b
)+2) * sizeof(*stats
));
155 coord_t best_move
= pass
;
156 int best_playouts
= -1;
162 for (int reply
= 0; reply
< reply_count
; reply
++) {
163 char *r
= gtp_replies
[reply
];
165 if (sscanf(r
, "=%d %d %d %d %d", &id
, &o
, &p
, &t
, &k
) != 5) continue;
167 *total_playouts
+= p
;
170 // Skip the rest of the firt line if any (allow future extensions)
174 struct move_stats2 s
;
175 while (r
&& sscanf(++r
, "%63s %d %f %d %f", move
, &s
.u
.playouts
,
176 &s
.u
.value
, &s
.amaf
.playouts
, &s
.amaf
.value
) == 5) {
177 coord_t
*c
= str2coord(move
, board_size(b
));
178 stats_add_result(&stats
[*c
].u
, s
.u
.value
, s
.u
.playouts
);
179 stats_add_result(&stats
[*c
].amaf
, s
.amaf
.value
, s
.amaf
.playouts
);
181 if (stats
[*c
].u
.playouts
> best_playouts
) {
182 best_playouts
= stats
[*c
].u
.playouts
;
189 *keep_looking
= keep
> reply_count
/ 2;
193 /* Set the args for the genmoves command. If stats is not null,
194 * append the stats from all slaves above min_playouts, except
195 * for pass and resign. args must have CMDS_SIZE bytes and
196 * upon return ends with an empty line.
197 * Keep this code in sync with uct_genmoves().
198 * slave_lock is held on entry and on return. */
200 genmoves_args(char *args
, struct board
*b
, enum stone color
, int played
,
201 struct time_info
*ti
, struct move_stats2
*stats
, int min_playouts
)
203 char *end
= args
+ CMDS_SIZE
;
204 char *s
= args
+ snprintf(args
, CMDS_SIZE
, "%s %d", stone2str(color
), played
);
206 if (ti
->dim
== TD_WALLTIME
) {
207 s
+= snprintf(s
, end
- s
, " %.3f %.3f %d %d",
208 ti
->len
.t
.main_time
, ti
->len
.t
.byoyomi_time
,
209 ti
->len
.t
.byoyomi_periods
, ti
->len
.t
.byoyomi_stones
);
211 s
+= snprintf(s
, end
- s
, "\n");
214 if (stats
[c
].u
.playouts
<= min_playouts
) continue;
215 s
+= snprintf(s
, end
- s
, "%s %d %.7f %d %.7f\n",
217 stats
[c
].u
.playouts
, stats
[c
].u
.value
,
218 stats
[c
].amaf
.playouts
, stats
[c
].amaf
.value
);
221 s
+= snprintf(s
, end
- s
, "\n");
224 /* Time control is mostly done by the slaves, so we use default values here. */
225 #define FUSEKI_END 20
226 #define YOSE_START 40
229 distributed_genmove(struct engine
*e
, struct board
*b
, struct time_info
*ti
,
230 enum stone color
, bool pass_all_alive
)
232 struct distributed
*dist
= e
->data
;
233 double now
= time_now();
236 char *cmd
= pass_all_alive
? "pachi-genmoves_cleanup" : "pachi-genmoves";
237 char args
[CMDS_SIZE
];
240 int played
, playouts
, threads
;
242 if (ti
->period
== TT_NULL
) *ti
= default_ti
;
243 struct time_stop stop
;
244 time_stop_conditions(ti
, b
, FUSEKI_END
, YOSE_START
, &stop
);
245 struct time_info saved_ti
= *ti
;
247 /* Send the first genmoves without stats. */
248 genmoves_args(args
, b
, color
, 0, ti
, NULL
, 0);
250 /* Combined move stats from all slaves, only for children
251 * of the root node, plus 2 for pass and resign. */
252 struct move_stats2
*stats
= alloca((board_size2(b
)+2) * sizeof(struct move_stats2
));
256 new_cmd(b
, cmd
, args
);
258 /* Loop until most slaves want to quit or time elapsed. */
261 /* Wait for just one slave to get stats as fresh as possible,
262 * or at most 100ms to check if we run out of time. */
263 get_replies(now
+ MAX_GENMOVES_WAIT
, 1);
265 if (ti
->dim
== TD_WALLTIME
)
266 time_sub(ti
, now
- start
, false);
269 best
= select_best_move(b
, stats
, &played
, &playouts
, &threads
, &keep_looking
);
271 if (!keep_looking
) break;
272 if (ti
->dim
== TD_WALLTIME
) {
273 if (now
- ti
->len
.t
.timer_start
>= stop
.worst
.time
) break;
275 if (played
>= stop
.worst
.playouts
) break;
279 char *coord
= coord2sstr(best
, b
);
280 snprintf(buf
, sizeof(buf
),
281 "temp winner is %s %s with score %1.4f (%d/%d games)"
282 " %d slaves %d threads\n",
283 stone2str(color
), coord
, get_value(stats
[best
].u
.value
, color
),
284 stats
[best
].u
.playouts
, playouts
, reply_count
, threads
);
285 logline(NULL
, "* ", buf
);
287 /* Send the command with the same gtp id, to avoid discarding
288 * a reply to a previous genmoves at the same move. */
289 /* Do not send ascii stats, slave now expects binary args. */
290 genmoves_args(args
, b
, color
, played
, ti
, NULL
, 0);
291 update_cmd(b
, cmd
, args
, false);
293 int replies
= reply_count
;
295 /* Do not subtract time spent twice (see gtp_parse). */
298 dist
->my_last_move
.color
= color
;
299 dist
->my_last_move
.coord
= best
;
300 dist
->my_last_stats
= stats
[best
].u
;
302 /* Tell the slaves to commit to the selected move, overwriting
303 * the last "pachi-genmoves" in the command history. */
304 char *coord
= coord2str(best
, b
);
305 snprintf(args
, sizeof(args
), "%s %s\n", stone2str(color
), coord
);
306 update_cmd(b
, "play", args
, true);
311 double time
= now
- first
+ 0.000001; /* avoid divide by zero */
312 snprintf(buf
, sizeof(buf
),
313 "GLOBAL WINNER is %s %s with score %1.4f (%d/%d games)\n"
314 "genmove %d games in %0.2fs %d slaves %d threads (%d games/s,"
315 " %d games/s/slave, %d games/s/thread)\n",
316 stone2str(color
), coord
, get_value(stats
[best
].u
.value
, color
),
317 stats
[best
].u
.playouts
, playouts
, played
, time
, replies
, threads
,
318 (int)(played
/time
), (int)(played
/time
/replies
),
319 (int)(played
/time
/threads
));
320 logline(NULL
, "* ", buf
);
323 return coord_copy(best
);
327 distributed_chat(struct engine
*e
, struct board
*b
, char *cmd
)
329 struct distributed
*dist
= e
->data
;
330 static char reply
[BSIZE
];
332 cmd
+= strspn(cmd
, " \n\t");
333 if (!strncasecmp(cmd
, "winrate", 7)) {
334 enum stone color
= dist
->my_last_move
.color
;
335 snprintf(reply
, BSIZE
, "In %d playouts at %d machines, %s %s can win with %.2f%% probability.",
336 dist
->my_last_stats
.playouts
, active_slaves
, stone2str(color
),
337 coord2sstr(dist
->my_last_move
.coord
, b
),
338 100 * get_value(dist
->my_last_stats
.value
, color
));
345 scmp(const void *p1
, const void *p2
)
347 return strcasecmp(*(char * const *)p1
, *(char * const *)p2
);
351 distributed_dead_group_list(struct engine
*e
, struct board
*b
, struct move_queue
*mq
)
355 new_cmd(b
, "final_status_list", "dead\n");
356 get_replies(time_now() + MAX_FAST_CMD_WAIT
, active_slaves
);
358 /* Find the most popular reply. */
359 qsort(gtp_replies
, reply_count
, sizeof(char *), scmp
);
363 for (int reply
= 1; reply
< reply_count
; reply
++) {
364 if (!strcmp(gtp_replies
[reply
], gtp_replies
[reply
-1])) {
369 if (count
> best_count
) {
375 /* Pick the first move of each line as group. */
376 char *dead
= gtp_replies
[best_reply
];
377 dead
= strchr(dead
, ' '); // skip "id "
378 while (dead
&& *++dead
!= '\n') {
379 coord_t
*c
= str2coord(dead
, board_size(b
));
382 dead
= strchr(dead
, '\n');
387 static struct distributed
*
388 distributed_state_init(char *arg
, struct board
*b
)
390 struct distributed
*dist
= calloc2(1, sizeof(struct distributed
));
392 dist
->max_slaves
= 100;
394 char *optspec
, *next
= arg
;
397 next
+= strcspn(next
, ",");
398 if (*next
) { *next
++ = 0; } else { *next
= 0; }
400 char *optname
= optspec
;
401 char *optval
= strchr(optspec
, '=');
402 if (optval
) *optval
++ = 0;
404 if (!strcasecmp(optname
, "slave_port") && optval
) {
405 dist
->slave_port
= strdup(optval
);
406 } else if (!strcasecmp(optname
, "proxy_port") && optval
) {
407 dist
->proxy_port
= strdup(optval
);
408 } else if (!strcasecmp(optname
, "max_slaves") && optval
) {
409 dist
->max_slaves
= atoi(optval
);
410 } else if (!strcasecmp(optname
, "slaves_quit")) {
411 dist
->slaves_quit
= !optval
|| atoi(optval
);
413 fprintf(stderr
, "distributed: Invalid engine argument %s or missing value\n", optname
);
418 gtp_replies
= calloc2(dist
->max_slaves
, sizeof(char *));
420 if (!dist
->slave_port
) {
421 fprintf(stderr
, "distributed: missing slave_port\n");
424 protocol_init(dist
->slave_port
, dist
->proxy_port
, dist
->max_slaves
);
429 engine_distributed_init(char *arg
, struct board
*b
)
431 struct distributed
*dist
= distributed_state_init(arg
, b
);
432 struct engine
*e
= calloc2(1, sizeof(struct engine
));
433 e
->name
= "Distributed Engine";
434 e
->comment
= "I'm playing the distributed engine. When I'm losing, I will resign, "
435 "if I think I win, I play until you pass. "
436 "Anyone can send me 'winrate' in private chat to get my assessment of the position.";
437 e
->notify
= distributed_notify
;
438 e
->genmove
= distributed_genmove
;
439 e
->dead_group_list
= distributed_dead_group_list
;
440 e
->chat
= distributed_chat
;
442 // Keep the threads and the open socket connections:
443 e
->keep_on_clear
= true;