uct_progress_text(): komi -> xkomi
[pachi/peepo.git] / distributed / protocol.c
blobc379c2a6f90cb053935067a26e4a8bb81feb8e99
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. */
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. */
13 #include <assert.h>
14 #include <stdio.h>
15 #include <pthread.h>
16 #include <ctype.h>
17 #include <unistd.h>
19 #define DEBUG
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"
29 /* All gtp commands for current game separated by \n */
30 static char gtp_cmds[CMDS_SIZE];
32 /* Latest gtp command sent to slaves. */
33 static char *gtp_cmd = NULL;
35 /* Slaves send gtp_cmd when cmd_count changes. */
36 static int cmd_count = 0;
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
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];
48 /* Number of active slave machines working for this master. */
49 int active_slaves = 0;
51 /* Number of replies to last gtp command already received. */
52 int reply_count = 0;
54 /* All replies to latest gtp command are in gtp_replies[0..reply_count-1]. */
55 char **gtp_replies;
58 struct buf_state **receive_queue;
59 int queue_length = 0;
60 int queue_age = 0;
61 static int queue_max_length;
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;
67 /* Condition signaled when a new gtp command is available. */
68 static pthread_cond_t cmd_cond = PTHREAD_COND_INITIALIZER;
70 /* Condition signaled when reply_count increases. */
71 static pthread_cond_t reply_cond = PTHREAD_COND_INITIALIZER;
73 /* Mutex protecting stderr. Must not be held at same time as slave_lock. */
74 static pthread_mutex_t log_lock = PTHREAD_MUTEX_INITIALIZER;
76 /* Absolute time when this program was started.
77 * For debugging only. */
78 static double start_time;
80 /* Default slave state. */
81 struct slave_state default_sstate;
84 /* Get exclusive access to the threads and commands state. */
85 void
86 protocol_lock(void)
88 pthread_mutex_lock(&slave_lock);
91 /* Release exclusive access to the threads and commands state. */
92 void
93 protocol_unlock(void)
95 pthread_mutex_unlock(&slave_lock);
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)
103 double now = time_now();
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';
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);
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)
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);
135 fclose(f);
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)
151 double start = time_now();
153 int reply_id = -1;
154 *reply = '\0';
155 if (!fgets(reply, CMDS_SIZE, f)) return -1;
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;
164 if (DEBUGV(s, 2))
165 logline(&client, "<<", reply);
166 if ((*reply == '=' || *reply == '?') && isdigit(reply[1]))
167 reply_id = atoi(reply+1);
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);
176 if (*line != '\n') return -1;
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;
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);
191 return size ? -1 : reply_id;
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)
205 assert(to_send && gtp_cmd && bin_buf && bin_size);
206 strncpy(buf, to_send, CMDS_SIZE);
207 bool resend = to_send != gtp_cmd;
209 pthread_mutex_unlock(&slave_lock);
211 if (DEBUGL(1) && resend)
212 logline(&sstate->client, "? ",
213 to_send == gtp_cmds ? "resend all\n" : "partial resend\n");
215 double start = time_now();
216 fputs(buf, f);
218 if (*bin_size)
219 fwrite(bin_buf, 1, *bin_size, f);
220 fflush(f);
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';
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);
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);
242 pthread_mutex_lock(&slave_lock);
243 return reply_id;
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)
253 if (cmd_id == -1) return gtp_cmds;
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;
259 int slot;
260 for (slot = 0; slot < MAX_CMDS_PER_MOVE; slot++) {
261 if (cmd_id == history[reply_move][slot].gtp_id) break;
263 if (slot == MAX_CMDS_PER_MOVE) return gtp_cmds;
265 char *next = history[reply_move][slot].next_cmd;
266 assert(next);
267 return next;
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)
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;
280 if (sstate->alloc_hook) sstate->alloc_hook(sstate);
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)
290 int newest = (sstate->newest_buf + 1) & (BUFFERS_PER_SLAVE - 1);
291 sstate->newest_buf = newest;
292 void *buf = sstate->b[newest].buf;
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);
302 int index = sstate->b[newest].queue_index;
303 if (index < 0) return buf;
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;
312 sstate->b[newest].queue_index = -1;
313 return buf;
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)
322 assert(queue_length < queue_max_length);
324 int newest = sstate->newest_buf;
325 assert(buf == sstate->b[newest].buf);
327 /* Update the buffer if necessary before making it
328 * available to other threads. */
329 if (sstate->insert_hook) sstate->insert_hook(buf, size);
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);
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++;
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)
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);
356 queue_length = 0;
357 queue_age++;
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)
370 /* Resend everything if slave returned an error. */
371 if (*reply != '=') {
372 *last_reply_id = -1;
373 return true;
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;
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;
389 if (bin_size) insert_buf(sstate, bin_reply, bin_size);
391 pthread_cond_signal(&reply_cond);
392 *last_reply_id = reply_id;
393 return false;
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)
409 int cmd_id = atoi(gtp_cmd);
410 void *buf = get_free_buf(sstate);
412 *bin_size = 0;
413 char *s = strchr(cmd, '@');
414 if (!s || !sstate->args_hook) return buf;
416 int size = sstate->args_hook(buf, sstate, cmd_id);
418 /* Check that the command is still valid. */
419 if (atoi(gtp_cmd) != cmd_id) return NULL;
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;
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)
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;
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;
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;
473 resend = process_reply(reply_id, buf, reply_buf, bin_buf, bin_size,
474 &last_reply_id, &reply_slot, sstate);
478 /* Minimimal check of slave identity. Close the file if error. */
479 static bool
480 is_pachi_slave(FILE *f, struct in_addr *client)
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;
493 return true;
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)
506 struct slave_state sstate = default_sstate;
507 sstate.thread_id = (long)arg;
509 assert(sstate.slave_sock >= 0);
510 char reply_buf[CMDS_SIZE];
511 bool resend = false;
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);
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);
524 if (!is_pachi_slave(f, &client)) continue;
526 if (!resend) slave_state_alloc(&sstate);
527 sstate.client = client;
529 pthread_mutex_lock(&slave_lock);
530 active_slaves++;
531 slave_loop(f, reply_buf, &sstate, resend);
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);
539 resend = true;
540 if (DEBUGL(2))
541 logline(&client, "= ", "lost slave\n");
542 fclose(f);
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)
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;
567 snprintf(gtp_cmd, gtp_cmds + CMDS_SIZE - gtp_cmd, "%d %s %s",
568 gtp_id, cmd, *args ? args : "\n");
569 cmd_count++;
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;
581 // Notify the slave threads about the new command.
582 pthread_cond_broadcast(&cmd_cond);
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)
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);
608 gtp_cmd += strlen(gtp_cmd);
611 // Let the slave threads send the new gtp command:
612 update_cmd(b, cmd, args, true);
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)
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);
633 if (reply_count == 0) continue;
634 if (reply_count >= min_replies || reply_count >= active_slaves) return;
635 if (time_now() >= time_limit) break;
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);
645 assert(reply_count > 0);
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
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)
658 start_time = time_now();
660 queue_max_length = max_slaves * MAX_GENMOVES_PER_SLAVE;
661 receive_queue = calloc2(queue_max_length, sizeof(*receive_queue));
663 default_sstate.slave_sock = port_listen(slave_port, max_slaves);
664 default_sstate.last_processed = -1;
666 for (int n = 0; n < BUFFERS_PER_SLAVE; n++) {
667 default_sstate.b[n].queue_index = -1;
670 pthread_t thread;
671 for (int id = 0; id < max_slaves; id++) {
672 pthread_create(&thread, NULL, slave_thread, (void *)(long)id);
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);