distributed/protocol.c

   1 /* The functions implementing the master-slave protocol of the
   2  * distributed engine are grouped here. They are independent
   3  * of the gtp protocol. See the comments at the top of distributed.c
   4  * for a general introduction to the distributed engine. */
   5
   6 /* The receive queue is an array of pointers to binary buffers.
   7  * These pointers are invalidated in one of two ways when a buffer
   8  * is recycled: (1) the queue age is increased when the queue is
   9  * emptied at a new move, (2) the pointer itself is set to NULL
  10  * immmediately, and stays so until at least the next queue age
  11  * increment. */
  12
  13 #include <assert.h>
  14 #include <stdio.h>
  15 #include <pthread.h>
  16 #include <ctype.h>
  17 #include <unistd.h>
  18
  19 #define DEBUG
  20
  21 #include "random.h"
  22 #include "timeinfo.h"
  23 #include "playout.h"
  24 #include "network.h"
  25 #include "debug.h"
  26 #include "distributed/distributed.h"
  27 #include "distributed/protocol.h"
  28
  29 /* All gtp commands for current game separated by \n */
  30 static char gtp_cmds[CMDS_SIZE];
  31
  32 /* Latest gtp command sent to slaves. */
  33 static char *gtp_cmd = NULL;
  34
  35 /* Slaves send gtp_cmd when cmd_count changes. */
  36 static int cmd_count = 0;
  37
  38 /* Remember at most 10 gtp ids per move: kgs-rules, boardsize, clear_board,
  39  * time_settings, komi, handicap, genmoves, play pass, play pass, final_status_list */
  40 #define MAX_CMDS_PER_MOVE 10
  41
  42 /* History of gtp commands sent for current game, indexed by move. */
  43 static struct cmd_history {
  44         int gtp_id;
  45         char *next_cmd;
  46 } history[MAX_GAMELEN][MAX_CMDS_PER_MOVE];
  47
  48 /* Number of active slave machines working for this master. */
  49 int active_slaves = 0;
  50
  51 /* Number of replies to last gtp command already received. */
  52 int reply_count = 0;
  53
  54 /* All replies to latest gtp command are in gtp_replies[0..reply_count-1]. */
  55 char **gtp_replies;
  56
  57
  58 struct buf_state **receive_queue;
  59 int queue_length = 0;
  60 int queue_age = 0;
  61 static int queue_max_length;
  62
  63 /* Mutex protecting all variables above. receive_queue may be
  64  * read without the lock but is only written with lock held. */
  65 static pthread_mutex_t slave_lock = PTHREAD_MUTEX_INITIALIZER;
  66
  67 /* Condition signaled when a new gtp command is available. */
  68 static pthread_cond_t cmd_cond = PTHREAD_COND_INITIALIZER;
  69
  70 /* Condition signaled when reply_count increases. */
  71 static pthread_cond_t reply_cond = PTHREAD_COND_INITIALIZER;
  72
  73 /* Mutex protecting stderr. Must not be held at same time as slave_lock. */
  74 static pthread_mutex_t log_lock = PTHREAD_MUTEX_INITIALIZER;
  75
  76 /* Absolute time when this program was started.
  77  * For debugging only. */
  78 static double start_time;
  79
  80 /* Default slave state. */
  81 struct slave_state default_sstate;
  82
  83
  84 /* Get exclusive access to the threads and commands state. */
  85 void
  86 protocol_lock(void)
  87 {
  88         pthread_mutex_lock(&slave_lock);
  89 }
  90
  91 /* Release exclusive access to the threads and commands state. */
  92 void
  93 protocol_unlock(void)
  94 {
  95         pthread_mutex_unlock(&slave_lock);
  96 }
  97
  98 /* Write the time, client address, prefix, and string s to stderr atomically.
  99  * s should end with a \n */
 100 void
 101 logline(struct in_addr *client, char *prefix, char *s)
 102 {
 103         double now = time_now();
 104
 105         char addr[INET_ADDRSTRLEN];
 106         if (client) {
 107 #ifdef _WIN32
 108                 strcpy(addr, inet_ntoa(*client));
 109 #else
 110                 inet_ntop(AF_INET, client, addr, sizeof(addr));
 111 #endif
 112         } else {
 113                 addr[0] = '\0';
 114         }
 115         pthread_mutex_lock(&log_lock);
 116         fprintf(stderr, "%s%15s %9.3f: %s", prefix, addr, now - start_time, s);
 117         pthread_mutex_unlock(&log_lock);
 118 }
 119
 120 /* Thread opening a connection on the given socket and copying input
 121  * from there to stderr. */
 122 static void *
 123 proxy_thread(void *arg)
 124 {
 125         int proxy_sock = (long)arg;
 126         assert(proxy_sock >= 0);
 127         for (;;) {
 128                 struct in_addr client;
 129                 int conn = open_server_connection(proxy_sock, &client);
 130                 FILE *f = fdopen(conn, "r");
 131                 char buf[BSIZE];
 132                 while (fgets(buf, BSIZE, f)) {
 133                         logline(&client, "< ", buf);
 134                 }
 135                 fclose(f);
 136         }
 137 }
 138
 139 /* Get a reply to one gtp command. Return the gtp command id,
 140  * or -1 if error. reply must have at least CMDS_SIZE bytes.
 141  * The ascii reply ends with an empty line; if the first line
 142  * contains "@size", a binary reply of size bytes follows the
 143  * empty line. @size is not standard gtp, it is only used
 144  * internally by Pachi for the genmoves command; it must be the
 145  * last parameter on the line.
 146  * *bin_size is the maximum size upon entry, actual size on return.
 147  * slave_lock is not held on either entry or exit of this function. */
 148 static int
 149 get_reply(FILE *f, struct in_addr client, char *reply, void *bin_reply, int *bin_size)
 150 {
 151         double start = time_now();
 152
 153         int reply_id = -1;
 154         *reply = '\0';
 155         if (!fgets(reply, CMDS_SIZE, f)) return -1;
 156
 157         /* Check for binary reply. */
 158         char *s = strchr(reply, '@');
 159         int size = 0;
 160         if (s) size = atoi(s+1);
 161         assert(size <= *bin_size);
 162         *bin_size = size;
 163
 164         if (DEBUGV(s, 2))
 165                 logline(&client, "<<", reply);
 166         if ((*reply == '=' || *reply == '?') && isdigit(reply[1]))
 167                 reply_id = atoi(reply+1);
 168
 169         /* Read the rest of the ascii reply */
 170         char *line = reply + strlen(reply);
 171         while (fgets(line, reply + CMDS_SIZE - line, f) && *line != '\n') {
 172                 if (DEBUGL(3))
 173                         logline(&client, "<<", line);
 174                 line += strlen(line);
 175         }
 176         if (*line != '\n') return -1;
 177
 178         /* Read the binary reply if any. */
 179         int len;
 180         while (size && (len = fread(bin_reply, 1, size, f)) > 0) {
 181                 bin_reply = (char *)bin_reply + len;
 182                 size -= len;
 183         }
 184         if (*bin_size && DEBUGVV(2)) {
 185                 char buf[1024];
 186                 snprintf(buf, sizeof(buf), "read reply %d+%d bytes in %.4fms\n",
 187                          (int)strlen(reply), *bin_size,
 188                          (time_now() - start)*1000);
 189                 logline(&client, "= ", buf);
 190         }
 191         return size ? -1 : reply_id;
 192 }
 193
 194 /* Send the gtp command to_send and get a reply from the slave machine.
 195  * Write the reply in buf which must have at least CMDS_SIZE bytes.
 196  * If *bin_size > 0, send bin_buf after the gtp command.
 197  * Return any binary reply in bin_buf and set its size in bin_size.
 198  * bin_buf is private to the slave and need not be copied.
 199  * Return the gtp command id, or -1 if error.
 200  * slave_lock is held on both entry and exit of this function. */
 201 static int
 202 send_command(char *to_send, void *bin_buf, int *bin_size,
 203              FILE *f, struct slave_state *sstate, char *buf)
 204 {
 205         assert(to_send && gtp_cmd && bin_buf && bin_size);
 206         strncpy(buf, to_send, CMDS_SIZE);
 207         bool resend = to_send != gtp_cmd;
 208
 209         pthread_mutex_unlock(&slave_lock);
 210
 211         if (DEBUGL(1) && resend)
 212                 logline(&sstate->client, "? ",
 213                         to_send == gtp_cmds ? "resend all\n" : "partial resend\n");
 214
 215         double start = time_now();
 216         fputs(buf, f);
 217
 218         if (*bin_size)
 219                 fwrite(bin_buf, 1, *bin_size, f);
 220         fflush(f);
 221
 222         if (DEBUGV(strchr(buf, '@'), 2)) {
 223                 double ms = (time_now() - start) * 1000.0;
 224                 if (!DEBUGL(3)) {
 225                         char *s = strchr(buf, '\n');
 226                         if (s) s[1] = '\0';
 227                 }
 228                 logline(&sstate->client, ">>", buf);
 229                 if (*bin_size) {
 230                         char b[1024];
 231                         snprintf(b, sizeof(b),
 232                                  "sent cmd %d+%d bytes in %.4fms\n",
 233                                  (int)strlen(buf), *bin_size, ms);
 234                         logline(&sstate->client, "= ", b);
 235                 }
 236         }
 237
 238         /* Reuse the buffers for the reply. */
 239         *bin_size = sstate->max_buf_size;
 240         int reply_id = get_reply(f, sstate->client, buf, bin_buf, bin_size);
 241
 242         pthread_mutex_lock(&slave_lock);
 243         return reply_id;
 244 }
 245
 246 /* Return the command sent after that with the given gtp id,
 247  * or gtp_cmds if the id wasn't used in this game. If a play command
 248  * has overwritten a genmoves command, return the play command.
 249  * slave_lock is held on both entry and exit of this function. */
 250 static char *
 251 next_command(int cmd_id)
 252 {
 253         if (cmd_id == -1) return gtp_cmds;
 254
 255         int last_id = atoi(gtp_cmd);
 256         int reply_move = move_number(cmd_id);
 257         if (reply_move > move_number(last_id)) return gtp_cmds;
 258
 259         int slot;
 260         for (slot = 0; slot < MAX_CMDS_PER_MOVE; slot++) {
 261                 if (cmd_id == history[reply_move][slot].gtp_id) break;
 262         }
 263         if (slot == MAX_CMDS_PER_MOVE) return gtp_cmds;
 264
 265         char *next = history[reply_move][slot].next_cmd;
 266         assert(next);
 267         return next;
 268 }
 269
 270 /* Allocate buffers for a slave thread. The state should have been
 271  * initialized already as a copy of the default slave state.
 272  * slave_lock is not held on either entry or exit of this function. */
 273 static void
 274 slave_state_alloc(struct slave_state *sstate)
 275 {
 276         for (int n = 0; n < BUFFERS_PER_SLAVE; n++) {
 277                 sstate->b[n].buf = malloc2(sstate->max_buf_size);
 278                 sstate->b[n].owner = sstate->thread_id;
 279         }
 280         if (sstate->alloc_hook) sstate->alloc_hook(sstate);
 281 }
 282
 283 /* Get a free binary buffer, first invalidating it in the receive
 284  * queue if necessary. In practice all buffers should be used
 285  * before they are invalidated, if BUFFERS_PER_SLAVE is large enough.
 286  * slave_lock is held on both entry and exit of this function. */
 287 static void *
 288 get_free_buf(struct slave_state *sstate)
 289 {
 290         int newest = (sstate->newest_buf + 1) & (BUFFERS_PER_SLAVE - 1);
 291         sstate->newest_buf = newest;
 292         void *buf = sstate->b[newest].buf;
 293
 294         if (DEBUGVV(7)) {
 295                 char b[1024];
 296                 snprintf(b, sizeof(b),
 297                          "get free %d index %d buf=%p age %d qlength %d\n", newest,
 298                          sstate->b[newest].queue_index, buf, queue_age, queue_length);
 299                 logline(&sstate->client, "? ", b);
 300         }
 301
 302         int index = sstate->b[newest].queue_index;
 303         if (index < 0) return buf;
 304
 305         /* Invalidate the buffer if the calling thread still owns its previous
 306          * entry in the receive queue. The entry may have been overwritten by
 307          * another thread, but only after a new move which invalidates the
 308          * entire receive queue. */
 309         if (receive_queue[index] && receive_queue[index]->owner == sstate->thread_id) {
 310                 receive_queue[index] = NULL;
 311         }
 312         sstate->b[newest].queue_index = -1;
 313         return buf;
 314 }
 315
 316 /* Insert a buffer in the receive queue. It should be the most
 317  * recent buffer allocated by the calling thread.
 318  * slave_lock is held on both entry and exit of this function. */
 319 static void
 320 insert_buf(struct slave_state *sstate, void *buf, int size)
 321 {
 322         assert(queue_length < queue_max_length);
 323
 324         int newest = sstate->newest_buf;
 325         assert(buf == sstate->b[newest].buf);
 326
 327         /* Update the buffer if necessary before making it
 328          * available to other threads. */
 329         if (sstate->insert_hook) sstate->insert_hook(buf, size);
 330
 331         if (DEBUGVV(7)) {
 332                 char b[1024];
 333                 snprintf(b, sizeof(b),
 334                          "insert newest %d age %d rq[%d]->%p owner %d\n",
 335                          newest, queue_age, queue_length, buf, sstate->thread_id);
 336                         logline(&sstate->client, "? ", b);
 337         }
 338         receive_queue[queue_length] = &sstate->b[newest];
 339         receive_queue[queue_length]->size = size;
 340         receive_queue[queue_length]->queue_index = queue_length;
 341         queue_length++;
 342 }
 343
 344 /* Clear the receive queue. The buffer pointers do not have to be cleared
 345  * here, this is done as each buffer is recycled.
 346  * slave_lock is held on both entry and exit of this function. */
 347 void
 348 clear_receive_queue(void)
 349 {
 350         if (DEBUGL(3)) {
 351                 char buf[1024];
 352                 snprintf(buf, sizeof(buf), "clear queue, old length %d age %d\n",
 353                          queue_length, queue_age);
 354                 logline(NULL, "? ", buf);
 355         }
 356         queue_length = 0;
 357         queue_age++;
 358 }
 359
 360 /* Process the reply received from a slave machine.
 361  * Copy the ascii part to reply_buf and insert the binary part
 362  * (if any) in the receive queue.
 363  * Return false if ok, true if the slave is out of sync.
 364  * slave_lock is held on both entry and exit of this function. */
 365 static bool
 366 process_reply(int reply_id, char *reply, char *reply_buf,
 367               void *bin_reply, int bin_size, int *last_reply_id,
 368               int *reply_slot, struct slave_state *sstate)
 369 {
 370         /* Resend everything if slave returned an error. */
 371         if (*reply != '=') {
 372                 *last_reply_id = -1;
 373                 return true;
 374         }
 375         /* Make sure we are still in sync. cmd_count may have
 376          * changed but the reply is valid as long as cmd_id didn't
 377          * change (this only occurs for consecutive genmoves). */
 378         int cmd_id = atoi(gtp_cmd);
 379         if (reply_id != cmd_id) {
 380                 *last_reply_id = reply_id;
 381                 return true;
 382         }
 383
 384         strncpy(reply_buf, reply, CMDS_SIZE);
 385         if (reply_id != *last_reply_id)
 386                 *reply_slot = reply_count++;
 387         gtp_replies[*reply_slot] = reply_buf;
 388
 389         if (bin_size) insert_buf(sstate, bin_reply, bin_size);
 390
 391         pthread_cond_signal(&reply_cond);
 392         *last_reply_id = reply_id;
 393         return false;
 394 }
 395
 396 /* Get the binary arg for the given command, and update the command
 397  * if necessary. For now, only genmoves has a binary argument, and
 398  * we return the best stats increments from all other slaves.
 399  * Set *bin_size to 0 if the command doesn't take binary arguments,
 400  * but still return a buffer, to be used for the reply.
 401  * Return NULL if the binary arg is obsolete by the time we have
 402  * finished computing it, because a new command is available.
 403  * This version only gets the buffer for the reply, to be completed
 404  * in future commits.
 405  * slave_lock is held on both entry and exit of this function. */
 406 void *
 407 get_binary_arg(struct slave_state *sstate, char *cmd, int cmd_size, int *bin_size)
 408 {
 409         int cmd_id = atoi(gtp_cmd);
 410         void *buf = get_free_buf(sstate);
 411
 412         *bin_size = 0;
 413         char *s = strchr(cmd, '@');
 414         if (!s || !sstate->args_hook) return buf;
 415
 416         int size = sstate->args_hook(buf, sstate, cmd_id);
 417
 418         /* Check that the command is still valid. */
 419         if (atoi(gtp_cmd) != cmd_id) return NULL;
 420
 421         /* Set the correct binary size for this slave.
 422          * cmd may have been overwritten with new parameters. */
 423         *bin_size = size;
 424         s = strchr(cmd, '@');
 425         assert(s);
 426         snprintf(s, cmd + cmd_size - s, "@%d\n", size);
 427         return buf;
 428 }
 429
 430 /* Main loop of a slave thread.
 431  * Send the current command to the slave machine and wait for a reply.
 432  * Resend command history if the slave machine is out of sync.
 433  * Returns when the connection with the slave machine is cut.
 434  * slave_lock is held on both entry and exit of this function. */
 435 static void
 436 slave_loop(FILE *f, char *reply_buf, struct slave_state *sstate, bool resend)
 437 {
 438         char *to_send;
 439         int last_cmd_count = 0;
 440         int last_reply_id = -1;
 441         int reply_slot = -1;
 442         for (;;) {
 443                 if (resend) {
 444                         /* Resend complete or partial history */
 445                         to_send = next_command(last_reply_id);
 446                 } else {
 447                         /* Wait for a new command. */
 448                         while (last_cmd_count == cmd_count)
 449                                 pthread_cond_wait(&cmd_cond, &slave_lock);
 450                         to_send = gtp_cmd;
 451                 }
 452
 453                 /* Command available, send it to slave machine.
 454                  * If slave was out of sync, send the history.
 455                  * But first get binary arguments if necessary. */
 456                 int bin_size = 0;
 457                 void *bin_buf = get_binary_arg(sstate, gtp_cmd,
 458                                                gtp_cmds + CMDS_SIZE - gtp_cmd,
 459                                                &bin_size);
 460                 /* Check that the command is still valid. */
 461                 resend = true;
 462                 if (!bin_buf) continue;
 463
 464                 /* Send the command and get the reply, which always ends with \n\n
 465                  * The slave machine sends "=id reply" or "?id reply"
 466                  * with id == cmd_id if it is in sync. */
 467                 last_cmd_count = cmd_count;
 468                 char buf[CMDS_SIZE];
 469                 int reply_id = send_command(to_send, bin_buf, &bin_size, f,
 470                                             sstate, buf);
 471                 if (reply_id == -1) return;
 472
 473                 resend = process_reply(reply_id, buf, reply_buf, bin_buf, bin_size,
 474                                        &last_reply_id, &reply_slot, sstate);
 475         }
 476 }
 477
 478 /* Minimimal check of slave identity. Close the file if error. */
 479 static bool
 480 is_pachi_slave(FILE *f, struct in_addr *client)
 481 {
 482         char buf[1024];
 483         fputs("name\n", f);
 484         if (!fgets(buf, sizeof(buf), f)
 485             || strncasecmp(buf, "= Pachi", 7)
 486             || !fgets(buf, sizeof(buf), f)
 487             || strcmp(buf, "\n")) {
 488                 logline(client, "? ", "bad slave\n");
 489                 fclose(f);
 490                 sleep(1); // avoid busy loop if error
 491                 return false;
 492         }
 493         return true;
 494 }
 495
 496 /* Thread sending gtp commands to one slave machine, and
 497  * reading replies. If a slave machine dies, this thread waits
 498  * for a connection from another slave.
 499  * The large buffers are allocated only once we get a first
 500  * connection, to avoid wasting memory if max_slaves is too large.
 501  * We do not invalidate the received buffers if a slave disconnects;
 502  * they are still useful for other slaves. */
 503 static void *
 504 slave_thread(void *arg)
 505 {
 506         struct slave_state sstate = default_sstate;
 507         sstate.thread_id = (long)arg;
 508
 509         assert(sstate.slave_sock >= 0);
 510         char reply_buf[CMDS_SIZE];
 511         bool resend = false;
 512
 513         for (;;) {
 514                 /* Wait for a connection from any slave. */
 515                 struct in_addr client;
 516                 int conn = open_server_connection(sstate.slave_sock, &client);
 517
 518                 FILE *f = fdopen(conn, "r+");
 519                 if (DEBUGL(2)) {
 520                         snprintf(reply_buf, sizeof(reply_buf),
 521                                  "new slave, id %d\n", sstate.thread_id);
 522                         logline(&client, "= ", reply_buf);
 523                 }
 524                 if (!is_pachi_slave(f, &client)) continue;
 525
 526                 if (!resend) slave_state_alloc(&sstate);
 527                 sstate.client = client;
 528
 529                 pthread_mutex_lock(&slave_lock);
 530                 active_slaves++;
 531                 slave_loop(f, reply_buf, &sstate, resend);
 532
 533                 assert(active_slaves > 0);
 534                 active_slaves--;
 535                 // Unblock main thread if it was waiting for this slave.
 536                 pthread_cond_signal(&reply_cond);
 537                 pthread_mutex_unlock(&slave_lock);
 538
 539                 resend = true;
 540                 if (DEBUGL(2))
 541                         logline(&client, "= ", "lost slave\n");
 542                 fclose(f);
 543         }
 544 }
 545
 546 /* Create a new gtp command for all slaves. The slave lock is held
 547  * upon entry and upon return, so the command will actually be
 548  * sent when the lock is released. The last command is overwritten
 549  * if gtp_cmd points to a non-empty string. cmd is a single word;
 550  * args has all arguments and is empty or has a trailing \n */
 551 void
 552 update_cmd(struct board *b, char *cmd, char *args, bool new_id)
 553 {
 554         assert(gtp_cmd);
 555         /* To make sure the slaves are in sync, we ignore the original id
 556          * and use the board number plus some random bits as gtp id. */
 557         static int gtp_id = -1;
 558         int moves = is_reset(cmd) ? 0 : b->moves;
 559         if (new_id) {
 560                 int prev_id = gtp_id;
 561                 do {
 562                         /* fast_random() is 16-bit only so the multiplication can't overflow. */
 563                         gtp_id = force_reply(moves + fast_random(65535) * DIST_GAMELEN);
 564                 } while (gtp_id == prev_id);
 565                 reply_count = 0;
 566         }
 567         snprintf(gtp_cmd, gtp_cmds + CMDS_SIZE - gtp_cmd, "%d %s %s",
 568                  gtp_id, cmd, *args ? args : "\n");
 569         cmd_count++;
 570
 571         /* Remember history for out-of-sync slaves. */
 572         static int slot = 0;
 573         static struct cmd_history *last = NULL;
 574         if (new_id) {
 575                 if (last) last->next_cmd = gtp_cmd;
 576                 slot = (slot + 1) % MAX_CMDS_PER_MOVE;
 577                 last = &history[moves][slot];
 578                 last->gtp_id = gtp_id;
 579                 last->next_cmd = NULL;
 580         }
 581         // Notify the slave threads about the new command.
 582         pthread_cond_broadcast(&cmd_cond);
 583 }
 584
 585 /* Update the command history, then create a new gtp command
 586  * for all slaves. The slave lock is held upon entry and
 587  * upon return, so the command will actually be sent when the
 588  * lock is released. cmd is a single word; args has all
 589  * arguments and is empty or has a trailing \n */
 590 void
 591 new_cmd(struct board *b, char *cmd, char *args)
 592 {
 593         // Clear the history when a new game starts:
 594         if (!gtp_cmd || is_gamestart(cmd)) {
 595                 gtp_cmd = gtp_cmds;
 596                 memset(history, 0, sizeof(history));
 597         } else {
 598                 /* Preserve command history for new slaves.
 599                  * To indicate that the slave should only reply to
 600                  * the last command we force the id of previous
 601                  * commands to be just the move number. */
 602                 int id = prevent_reply(atoi(gtp_cmd));
 603                 int len = strspn(gtp_cmd, "0123456789");
 604                 char buf[32];
 605                 snprintf(buf, sizeof(buf), "%0*d", len, id);
 606                 memcpy(gtp_cmd, buf, len);
 607
 608                 gtp_cmd += strlen(gtp_cmd);
 609         }
 610
 611         // Let the slave threads send the new gtp command:
 612         update_cmd(b, cmd, args, true);
 613 }
 614
 615 /* Wait for at least one new reply. Return when at least
 616  * min_replies slaves have already replied, or when the
 617  * given absolute time is passed.
 618  * The replies are returned in gtp_replies[0..reply_count-1]
 619  * slave_lock is held on entry and on return. */
 620 void
 621 get_replies(double time_limit, int min_replies)
 622 {
 623         for (;;) {
 624                 if (reply_count > 0) {
 625                         struct timespec ts;
 626                         double sec;
 627                         ts.tv_nsec = (int)(modf(time_limit, &sec)*1000000000.0);
 628                         ts.tv_sec = (int)sec;
 629                         pthread_cond_timedwait(&reply_cond, &slave_lock, &ts);
 630                 } else {
 631                         pthread_cond_wait(&reply_cond, &slave_lock);
 632                 }
 633                 if (reply_count == 0) continue;
 634                 if (reply_count >= min_replies || reply_count >= active_slaves) return;
 635                 if (time_now() >= time_limit) break;
 636         }
 637         if (DEBUGL(1)) {
 638                 char buf[1024];
 639                 snprintf(buf, sizeof(buf),
 640                          "get_replies timeout %.3f >= %.3f, replies %d < min %d, active %d\n",
 641                          time_now() - start_time, time_limit - start_time,
 642                          reply_count, min_replies, active_slaves);
 643                 logline(NULL, "? ", buf);
 644         }
 645         assert(reply_count > 0);
 646 }
 647
 648 /* In a 5mn move with at least 5ms per genmoves we get at most
 649  * 300*200=60000 genmoves per slave. */
 650 #define MAX_GENMOVES_PER_SLAVE 60000
 651
 652 /* Allocate the receive queue, and create the slave and proxy threads.
 653  * max_buf_size and the merge-related fields of default_sstate must
 654  * already be initialized. */
 655 void
 656 protocol_init(char *slave_port, char *proxy_port, int max_slaves)
 657 {
 658         start_time = time_now();
 659
 660         queue_max_length = max_slaves * MAX_GENMOVES_PER_SLAVE;
 661         receive_queue = calloc2(queue_max_length, sizeof(*receive_queue));
 662
 663         default_sstate.slave_sock = port_listen(slave_port, max_slaves);
 664         default_sstate.last_processed = -1;
 665
 666         for (int n = 0; n < BUFFERS_PER_SLAVE; n++) {
 667                 default_sstate.b[n].queue_index = -1;
 668         }
 669
 670         pthread_t thread;
 671         for (int id = 0; id < max_slaves; id++) {
 672                 pthread_create(&thread, NULL, slave_thread, (void *)(long)id);
 673         }
 674
 675         if (proxy_port) {
 676                 int proxy_sock = port_listen(proxy_port, max_slaves);
 677                 for (int id = 0; id < max_slaves; id++) {
 678                         pthread_create(&thread, NULL, proxy_thread, (void *)(long)proxy_sock);
 679                 }
 680         }
 681 }