distributed/distributed.c

   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. */
   9
  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. */
  18
  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. */
  30
  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.) */
  39
  40 /* The master-slave protocol has fault tolerance. If a slave is
  41  * out of sync, the master sends it the appropriate command history. */
  42
  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
  55  */
  56
  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
  67  */
  68
  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>
  76
  77 #define DEBUG
  78
  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"
  89
  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;
 102 };
 103
 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 },
 111 };
 112
 113 #define get_value(value, color) \
 114         ((color) == S_BLACK ? (value) : 1 - (value))
 115
 116
 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
 120
 121 /* Maximum time (seconds) to wait for answers to genmoves. */
 122 #define MAX_GENMOVES_WAIT 0.1 /* 100 ms */
 123
 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 */
 127
 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)
 133 {
 134         /* Special case for pass and resign. */
 135         if (path < 0) return coord2sstr((coord_t)path, b);
 136
 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);
 148         }
 149         if (s != b2) s[-1] = '\0';
 150         return b2;
 151 }
 152
 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)
 158 {
 159         struct distributed *dist = e->data;
 160
 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")
 169
 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;
 176
 177         protocol_lock();
 178
 179         // Create a new command to be sent by the slave threads.
 180         new_cmd(b, cmd, args);
 181
 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);
 189
 190         protocol_unlock();
 191
 192         // At the beginning wait even more for late slaves.
 193         if (b->moves == 0) sleep(1);
 194         return P_OK;
 195 }
 196
 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
 203 };
 204
 205 static void
 206 large_stats_add_result(struct large_stats *s, floating_t result, long playouts)
 207 {
 208         s->playouts += playouts;
 209         s->value += (result - s->value) * playouts / s->playouts;
 210 }
 211
 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)
 225 {
 226         assert(reply_count > 0);
 227
 228         /* +2 for pass and resign */
 229         memset(stats-2, 0, (board_size2(b)+2) * sizeof(*stats));
 230
 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;
 237
 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');
 248
 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);
 254
 255                         large_stats_add_result(&stats[c], s.value, (long)s.playouts);
 256
 257                         if (stats[c].playouts > best_playouts) {
 258                                 best_playouts = stats[c].playouts;
 259                                 best_move = c;
 260                         }
 261                         r = strchr(r, '\n');
 262                 }
 263         }
 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;
 268 }
 269
 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)
 280 {
 281         char *end = args + CMDS_SIZE;
 282         char *s = args + snprintf(args, CMDS_SIZE, "%s %d", stone2str(color), played);
 283
 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);
 288         }
 289         s += snprintf(s, end - s, binary_args ? " @0\n" : "\n");
 290 }
 291
 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
 296
 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)
 301 {
 302         struct distributed *dist = e->data;
 303         double now = time_now();
 304         double first = now;
 305         char buf[BSIZE]; // debug only
 306
 307         char *cmd = pass_all_alive ? "pachi-genmoves_cleanup" : "pachi-genmoves";
 308         char args[CMDS_SIZE];
 309
 310         coord_t best;
 311         int played, playouts, threads;
 312
 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;
 317
 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];
 322
 323         protocol_lock();
 324         clear_receive_queue();
 325
 326         /* Send the first genmoves without stats. */
 327         genmoves_args(args, color, 0, ti, false);
 328         new_cmd(b, cmd, args);
 329
 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);
 340
 341                 bool keep_looking;
 342                 best = select_best_move(b, stats, &played, &playouts, &threads, &keep_looking);
 343
 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;
 349                 }
 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);
 358                 }
 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);
 363         }
 364         int replies = reply_count;
 365
 366         /* Do not subtract time spent twice (see gtp_parse). */
 367         *ti = saved_ti;
 368
 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;
 375
 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();
 384
 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);
 396         }
 397         if (DEBUGL(3)) {
 398                 int total_hnodes = replies * (1 << dist->stats_hbits);
 399                 merge_print_stats(total_hnodes);
 400         }
 401         return coord_copy(best);
 402 }
 403
 404 static char *
 405 distributed_chat(struct engine *e, struct board *b, bool opponent, char *from, char *cmd)
 406 {
 407         struct distributed *dist = e->data;
 408         double winrate = get_value(dist->my_last_stats.value, dist->my_last_move.color);
 409
 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);
 412 }
 413
 414 static int
 415 scmp(const void *p1, const void *p2)
 416 {
 417         return strcasecmp(*(char * const *)p1, *(char * const *)p2);
 418 }
 419
 420 static void
 421 distributed_dead_group_list(struct engine *e, struct board *b, struct move_queue *mq)
 422 {
 423         protocol_lock();
 424
 425         new_cmd(b, "final_status_list", "dead\n");
 426         get_replies(time_now() + MAX_FAST_CMD_WAIT, active_slaves);
 427
 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;
 438                 }
 439                 if (count > best_count) {
 440                         best_count = count;
 441                         best_reply = reply;
 442                 }
 443         }
 444
 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');
 451         }
 452         protocol_unlock();
 453 }
 454
 455 static struct distributed *
 456 distributed_state_init(char *arg, struct board *b)
 457 {
 458         struct distributed *dist = calloc2(1, sizeof(struct distributed));
 459
 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; }
 469
 470                         char *optname = optspec;
 471                         char *optval = strchr(optspec, '=');
 472                         if (optval) *optval++ = 0;
 473
 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);
 491                         }
 492                 }
 493         }
 494
 495         gtp_replies = calloc2(dist->max_slaves, sizeof(char *));
 496
 497         if (!dist->slave_port) {
 498                 fprintf(stderr, "distributed: missing slave_port\n");
 499                 exit(1);
 500         }
 501
 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);
 504
 505         return dist;
 506 }
 507
 508 struct engine *
 509 engine_distributed_init(char *arg, struct board *b)
 510 {
 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;
 524
 525         return e;
 526 }