Distributed engine: Define is_pachi_slave() and avoid busy loop if error
[pachi.git] / distributed / protocol.c
blobda23b5f152d7ff344ebbe598fd28760ce19b6ced
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 inet_ntop(AF_INET, client, addr, sizeof(addr));
108 } else {
109 addr[0] = '\0';
111 pthread_mutex_lock(&log_lock);
112 fprintf(stderr, "%s%15s %9.3f: %s", prefix, addr, now - start_time, s);
113 pthread_mutex_unlock(&log_lock);
116 /* Thread opening a connection on the given socket and copying input
117 * from there to stderr. */
118 static void *
119 proxy_thread(void *arg)
121 int proxy_sock = (long)arg;
122 assert(proxy_sock >= 0);
123 for (;;) {
124 struct in_addr client;
125 int conn = open_server_connection(proxy_sock, &client);
126 FILE *f = fdopen(conn, "r");
127 char buf[BSIZE];
128 while (fgets(buf, BSIZE, f)) {
129 logline(&client, "< ", buf);
131 fclose(f);
135 /* Get a reply to one gtp command. Return the gtp command id,
136 * or -1 if error. reply must have at least CMDS_SIZE bytes.
137 * The ascii reply ends with an empty line; if the first line
138 * contains "@size", a binary reply of size bytes follows the
139 * empty line. @size is not standard gtp, it is only used
140 * internally by Pachi for the genmoves command; it must be the
141 * last parameter on the line.
142 * *bin_size is the maximum size upon entry, actual size on return.
143 * slave_lock is not held on either entry or exit of this function. */
144 static int
145 get_reply(FILE *f, struct in_addr client, char *reply, void *bin_reply, int *bin_size)
147 int reply_id = -1;
148 *reply = '\0';
149 if (!fgets(reply, CMDS_SIZE, f)) return -1;
151 /* Check for binary reply. */
152 char *s = strchr(reply, '@');
153 int size = 0;
154 if (s) size = atoi(s+1);
155 assert(size <= *bin_size);
156 *bin_size = size;
158 if (DEBUGV(s, 2))
159 logline(&client, "<<", reply);
160 if ((*reply == '=' || *reply == '?') && isdigit(reply[1]))
161 reply_id = atoi(reply+1);
163 /* Read the rest of the ascii reply */
164 char *line = reply + strlen(reply);
165 while (fgets(line, reply + CMDS_SIZE - line, f) && *line != '\n') {
166 if (DEBUGL(3))
167 logline(&client, "<<", line);
168 line += strlen(line);
170 if (*line != '\n') return -1;
172 /* Read the binary reply if any. */
173 double start = time_now();
174 int len;
175 while (size && (len = fread(bin_reply, 1, size, f)) > 0) {
176 bin_reply = (char *)bin_reply + len;
177 size -= len;
179 if (*bin_size && DEBUGVV(2)) {
180 char buf[1024];
181 snprintf(buf, sizeof(buf), "read reply %d bytes in %.4fms\n", *bin_size,
182 (time_now() - start)*1000);
183 logline(&client, "= ", buf);
185 return size ? -1 : reply_id;
188 /* Send the gtp command to_send and get a reply from the slave machine.
189 * Write the reply in buf which must have at least CMDS_SIZE bytes.
190 * If *bin_size > 0, send bin_buf after the gtp command.
191 * Return any binary reply in bin_buf and set its size in bin_size.
192 * bin_buf is private to the slave and need not be copied.
193 * Return the gtp command id, or -1 if error.
194 * slave_lock is held on both entry and exit of this function. */
195 static int
196 send_command(char *to_send, void *bin_buf, int *bin_size,
197 FILE *f, struct slave_state *sstate, char *buf)
199 assert(to_send && gtp_cmd && bin_buf && bin_size);
200 strncpy(buf, to_send, CMDS_SIZE);
201 bool resend = to_send != gtp_cmd;
203 pthread_mutex_unlock(&slave_lock);
205 if (DEBUGL(1) && resend)
206 logline(&sstate->client, "? ",
207 to_send == gtp_cmds ? "resend all\n" : "partial resend\n");
208 fputs(buf, f);
210 double start = time_now();
211 if (*bin_size)
212 fwrite(bin_buf, 1, *bin_size, f);
213 fflush(f);
215 if (DEBUGV(strchr(buf, '@'), 2)) {
216 double ms = (time_now() - start) * 1000.0;
217 if (!DEBUGL(3)) {
218 char *s = strchr(buf, '\n');
219 if (s) s[1] = '\0';
221 logline(&sstate->client, ">>", buf);
222 if (*bin_size) {
223 char b[1024];
224 snprintf(b, sizeof(b),
225 "sent args %d bytes in %.4fms\n", *bin_size, ms);
226 logline(&sstate->client, "= ", b);
230 /* Reuse the buffers for the reply. */
231 *bin_size = sstate->max_buf_size;
232 int reply_id = get_reply(f, sstate->client, buf, bin_buf, bin_size);
234 pthread_mutex_lock(&slave_lock);
235 return reply_id;
238 /* Return the command sent after that with the given gtp id,
239 * or gtp_cmds if the id wasn't used in this game. If a play command
240 * has overwritten a genmoves command, return the play command.
241 * slave_lock is held on both entry and exit of this function. */
242 static char *
243 next_command(int cmd_id)
245 if (cmd_id == -1) return gtp_cmds;
247 int last_id = atoi(gtp_cmd);
248 int reply_move = move_number(cmd_id);
249 if (reply_move > move_number(last_id)) return gtp_cmds;
251 int slot;
252 for (slot = 0; slot < MAX_CMDS_PER_MOVE; slot++) {
253 if (cmd_id == history[reply_move][slot].gtp_id) break;
255 if (slot == MAX_CMDS_PER_MOVE) return gtp_cmds;
257 char *next = history[reply_move][slot].next_cmd;
258 assert(next);
259 return next;
262 /* Allocate buffers for a slave thread. The state should have been
263 * initialized already as a copy of the default slave state.
264 * slave_lock is not held on either entry or exit of this function. */
265 static void
266 slave_state_alloc(struct slave_state *sstate)
268 for (int n = 0; n < BUFFERS_PER_SLAVE; n++) {
269 sstate->b[n].buf = malloc2(sstate->max_buf_size);
270 sstate->b[n].owner = sstate->thread_id;
272 if (sstate->alloc_hook) sstate->alloc_hook(sstate);
275 /* Get a free binary buffer, first invalidating it in the receive
276 * queue if necessary. In practice all buffers should be used
277 * before they are invalidated, if BUFFERS_PER_SLAVE is large enough.
278 * slave_lock is held on both entry and exit of this function. */
279 static void *
280 get_free_buf(struct slave_state *sstate)
282 int newest = (sstate->newest_buf + 1) & (BUFFERS_PER_SLAVE - 1);
283 sstate->newest_buf = newest;
284 void *buf = sstate->b[newest].buf;
286 if (DEBUGVV(7)) {
287 char b[1024];
288 snprintf(b, sizeof(b),
289 "get free %d index %d buf=%p age %d qlength %d\n", newest,
290 sstate->b[newest].queue_index, buf, queue_age, queue_length);
291 logline(&sstate->client, "? ", b);
294 int index = sstate->b[newest].queue_index;
295 if (index < 0) return buf;
297 /* Invalidate the buffer if the calling thread still owns its previous
298 * entry in the receive queue. The entry may have been overwritten by
299 * another thread, but only after a new move which invalidates the
300 * entire receive queue. */
301 if (receive_queue[index] && receive_queue[index]->owner == sstate->thread_id) {
302 receive_queue[index] = NULL;
304 sstate->b[newest].queue_index = -1;
305 return buf;
308 /* Insert a buffer in the receive queue. It should be the most
309 * recent buffer allocated by the calling thread.
310 * slave_lock is held on both entry and exit of this function. */
311 static void
312 insert_buf(struct slave_state *sstate, void *buf, int size)
314 assert(queue_length < queue_max_length);
316 int newest = sstate->newest_buf;
317 assert(buf == sstate->b[newest].buf);
319 /* Update the buffer if necessary before making it
320 * available to other threads. */
321 if (sstate->insert_hook) sstate->insert_hook(buf, size);
323 if (DEBUGVV(7)) {
324 char b[1024];
325 snprintf(b, sizeof(b),
326 "insert newest %d age %d rq[%d]->%p owner %d\n",
327 newest, queue_age, queue_length, buf, sstate->thread_id);
328 logline(&sstate->client, "? ", b);
330 receive_queue[queue_length] = &sstate->b[newest];
331 receive_queue[queue_length]->size = size;
332 receive_queue[queue_length]->queue_index = queue_length;
333 queue_length++;
336 /* Clear the receive queue. The buffer pointers do not have to be cleared
337 * here, this is done as each buffer is recycled.
338 * slave_lock is held on both entry and exit of this function. */
339 void
340 clear_receive_queue(void)
342 if (DEBUGL(3)) {
343 char buf[1024];
344 snprintf(buf, sizeof(buf), "clear queue, old length %d age %d\n",
345 queue_length, queue_age);
346 logline(NULL, "? ", buf);
348 queue_length = 0;
349 queue_age++;
352 /* Process the reply received from a slave machine.
353 * Copy the ascii part to reply_buf and insert the binary part
354 * (if any) in the receive queue.
355 * Return false if ok, true if the slave is out of sync.
356 * slave_lock is held on both entry and exit of this function. */
357 static bool
358 process_reply(int reply_id, char *reply, char *reply_buf,
359 void *bin_reply, int bin_size, int *last_reply_id,
360 int *reply_slot, struct slave_state *sstate)
362 /* Resend everything if slave returned an error. */
363 if (*reply != '=') {
364 *last_reply_id = -1;
365 return true;
367 /* Make sure we are still in sync. cmd_count may have
368 * changed but the reply is valid as long as cmd_id didn't
369 * change (this only occurs for consecutive genmoves). */
370 int cmd_id = atoi(gtp_cmd);
371 if (reply_id != cmd_id) {
372 *last_reply_id = reply_id;
373 return true;
376 strncpy(reply_buf, reply, CMDS_SIZE);
377 if (reply_id != *last_reply_id)
378 *reply_slot = reply_count++;
379 gtp_replies[*reply_slot] = reply_buf;
381 if (bin_size) insert_buf(sstate, bin_reply, bin_size);
383 pthread_cond_signal(&reply_cond);
384 *last_reply_id = reply_id;
385 return false;
388 /* Get the binary arg for the given command, and update the command
389 * if necessary. For now, only genmoves has a binary argument, and
390 * we return the best stats increments from all other slaves.
391 * Set *bin_size to 0 if the command doesn't take binary arguments,
392 * but still return a buffer, to be used for the reply.
393 * Return NULL if the binary arg is obsolete by the time we have
394 * finished computing it, because a new command is available.
395 * This version only gets the buffer for the reply, to be completed
396 * in future commits.
397 * slave_lock is held on both entry and exit of this function. */
398 void *
399 get_binary_arg(struct slave_state *sstate, char *cmd, int cmd_size, int *bin_size)
401 int cmd_id = atoi(gtp_cmd);
402 void *buf = get_free_buf(sstate);
404 *bin_size = 0;
405 char *s = strchr(cmd, '@');
406 if (!s || !sstate->args_hook) return buf;
408 int size = sstate->args_hook(buf, sstate, cmd_id);
410 /* Check that the command is still valid. */
411 if (atoi(gtp_cmd) != cmd_id) return NULL;
413 /* Set the correct binary size for this slave.
414 * cmd may have been overwritten with new parameters. */
415 *bin_size = size;
416 s = strchr(cmd, '@');
417 assert(s);
418 snprintf(s, cmd + cmd_size - s, "@%d\n", size);
419 return buf;
422 /* Main loop of a slave thread.
423 * Send the current command to the slave machine and wait for a reply.
424 * Resend command history if the slave machine is out of sync.
425 * Returns when the connection with the slave machine is cut.
426 * slave_lock is held on both entry and exit of this function. */
427 static void
428 slave_loop(FILE *f, char *reply_buf, struct slave_state *sstate, bool resend)
430 char *to_send;
431 int last_cmd_count = 0;
432 int last_reply_id = -1;
433 int reply_slot = -1;
434 for (;;) {
435 if (resend) {
436 /* Resend complete or partial history */
437 to_send = next_command(last_reply_id);
438 } else {
439 /* Wait for a new command. */
440 while (last_cmd_count == cmd_count)
441 pthread_cond_wait(&cmd_cond, &slave_lock);
442 to_send = gtp_cmd;
445 /* Command available, send it to slave machine.
446 * If slave was out of sync, send the history.
447 * But first get binary arguments if necessary. */
448 int bin_size = 0;
449 void *bin_buf = get_binary_arg(sstate, gtp_cmd,
450 gtp_cmds + CMDS_SIZE - gtp_cmd,
451 &bin_size);
452 /* Check that the command is still valid. */
453 resend = true;
454 if (!bin_buf) continue;
456 /* Send the command and get the reply, which always ends with \n\n
457 * The slave machine sends "=id reply" or "?id reply"
458 * with id == cmd_id if it is in sync. */
459 last_cmd_count = cmd_count;
460 char buf[CMDS_SIZE];
461 int reply_id = send_command(to_send, bin_buf, &bin_size, f,
462 sstate, buf);
463 if (reply_id == -1) return;
465 resend = process_reply(reply_id, buf, reply_buf, bin_buf, bin_size,
466 &last_reply_id, &reply_slot, sstate);
470 /* Minimimal check of slave identity. Close the file if error. */
471 static bool
472 is_pachi_slave(FILE *f, struct in_addr *client)
474 char buf[1024];
475 fputs("name\n", f);
476 if (!fgets(buf, sizeof(buf), f)
477 || strncasecmp(buf, "= Pachi", 7)
478 || !fgets(buf, sizeof(buf), f)
479 || strcmp(buf, "\n")) {
480 logline(client, "? ", "bad slave\n");
481 fclose(f);
482 sleep(1); // avoid busy loop if error
483 return false;
485 return true;
488 /* Thread sending gtp commands to one slave machine, and
489 * reading replies. If a slave machine dies, this thread waits
490 * for a connection from another slave.
491 * The large buffers are allocated only once we get a first
492 * connection, to avoid wasting memory if max_slaves is too large.
493 * We do not invalidate the received buffers if a slave disconnects;
494 * they are still useful for other slaves. */
495 static void *
496 slave_thread(void *arg)
498 struct slave_state sstate = default_sstate;
499 sstate.thread_id = (long)arg;
501 assert(sstate.slave_sock >= 0);
502 char reply_buf[CMDS_SIZE];
503 bool resend = false;
505 for (;;) {
506 /* Wait for a connection from any slave. */
507 struct in_addr client;
508 int conn = open_server_connection(sstate.slave_sock, &client);
510 FILE *f = fdopen(conn, "r+");
511 if (DEBUGL(2)) {
512 snprintf(reply_buf, sizeof(reply_buf),
513 "new slave, id %d\n", sstate.thread_id);
514 logline(&client, "= ", reply_buf);
516 if (!is_pachi_slave(f, &client)) continue;
518 if (!resend) slave_state_alloc(&sstate);
519 sstate.client = client;
521 pthread_mutex_lock(&slave_lock);
522 active_slaves++;
523 slave_loop(f, reply_buf, &sstate, resend);
525 assert(active_slaves > 0);
526 active_slaves--;
527 // Unblock main thread if it was waiting for this slave.
528 pthread_cond_signal(&reply_cond);
529 pthread_mutex_unlock(&slave_lock);
531 resend = true;
532 if (DEBUGL(2))
533 logline(&client, "= ", "lost slave\n");
534 fclose(f);
538 /* Create a new gtp command for all slaves. The slave lock is held
539 * upon entry and upon return, so the command will actually be
540 * sent when the lock is released. The last command is overwritten
541 * if gtp_cmd points to a non-empty string. cmd is a single word;
542 * args has all arguments and is empty or has a trailing \n */
543 void
544 update_cmd(struct board *b, char *cmd, char *args, bool new_id)
546 assert(gtp_cmd);
547 /* To make sure the slaves are in sync, we ignore the original id
548 * and use the board number plus some random bits as gtp id. */
549 static int gtp_id = -1;
550 int moves = is_reset(cmd) ? 0 : b->moves;
551 if (new_id) {
552 int prev_id = gtp_id;
553 do {
554 /* fast_random() is 16-bit only so the multiplication can't overflow. */
555 gtp_id = force_reply(moves + fast_random(65535) * DIST_GAMELEN);
556 } while (gtp_id == prev_id);
557 reply_count = 0;
559 snprintf(gtp_cmd, gtp_cmds + CMDS_SIZE - gtp_cmd, "%d %s %s",
560 gtp_id, cmd, *args ? args : "\n");
561 cmd_count++;
563 /* Remember history for out-of-sync slaves. */
564 static int slot = 0;
565 static struct cmd_history *last = NULL;
566 if (new_id) {
567 if (last) last->next_cmd = gtp_cmd;
568 slot = (slot + 1) % MAX_CMDS_PER_MOVE;
569 last = &history[moves][slot];
570 last->gtp_id = gtp_id;
571 last->next_cmd = NULL;
573 // Notify the slave threads about the new command.
574 pthread_cond_broadcast(&cmd_cond);
577 /* Update the command history, then create a new gtp command
578 * for all slaves. The slave lock is held upon entry and
579 * upon return, so the command will actually be sent when the
580 * lock is released. cmd is a single word; args has all
581 * arguments and is empty or has a trailing \n */
582 void
583 new_cmd(struct board *b, char *cmd, char *args)
585 // Clear the history when a new game starts:
586 if (!gtp_cmd || is_gamestart(cmd)) {
587 gtp_cmd = gtp_cmds;
588 memset(history, 0, sizeof(history));
589 } else {
590 /* Preserve command history for new slaves.
591 * To indicate that the slave should only reply to
592 * the last command we force the id of previous
593 * commands to be just the move number. */
594 int id = prevent_reply(atoi(gtp_cmd));
595 int len = strspn(gtp_cmd, "0123456789");
596 char buf[32];
597 snprintf(buf, sizeof(buf), "%0*d", len, id);
598 memcpy(gtp_cmd, buf, len);
600 gtp_cmd += strlen(gtp_cmd);
603 // Let the slave threads send the new gtp command:
604 update_cmd(b, cmd, args, true);
607 /* Wait for at least one new reply. Return when at least
608 * min_replies slaves have already replied, or when the
609 * given absolute time is passed.
610 * The replies are returned in gtp_replies[0..reply_count-1]
611 * slave_lock is held on entry and on return. */
612 void
613 get_replies(double time_limit, int min_replies)
615 for (;;) {
616 if (reply_count > 0) {
617 struct timespec ts;
618 double sec;
619 ts.tv_nsec = (int)(modf(time_limit, &sec)*1000000000.0);
620 ts.tv_sec = (int)sec;
621 pthread_cond_timedwait(&reply_cond, &slave_lock, &ts);
622 } else {
623 pthread_cond_wait(&reply_cond, &slave_lock);
625 if (reply_count == 0) continue;
626 if (reply_count >= min_replies || reply_count >= active_slaves) return;
627 if (time_now() >= time_limit) break;
629 if (DEBUGL(1)) {
630 char buf[1024];
631 snprintf(buf, sizeof(buf),
632 "get_replies timeout %.3f >= %.3f, replies %d < min %d, active %d\n",
633 time_now() - start_time, time_limit - start_time,
634 reply_count, min_replies, active_slaves);
635 logline(NULL, "? ", buf);
637 assert(reply_count > 0);
640 /* In a 30s move with at least 5ms per genmoves we get at most
641 * 6000 genmoves per slave. */
642 #define MAX_GENMOVES_PER_SLAVE 6000
644 /* Allocate the receive queue, and create the slave and proxy threads.
645 * max_buf_size and the merge-related fields of default_sstate must
646 * already be initialized. */
647 void
648 protocol_init(char *slave_port, char *proxy_port, int max_slaves)
650 start_time = time_now();
652 queue_max_length = max_slaves * MAX_GENMOVES_PER_SLAVE;
653 receive_queue = calloc2(queue_max_length, sizeof(*receive_queue));
655 default_sstate.slave_sock = port_listen(slave_port, max_slaves);
656 default_sstate.last_processed = -1;
658 for (int n = 0; n < BUFFERS_PER_SLAVE; n++) {
659 default_sstate.b[n].queue_index = -1;
662 pthread_t thread;
663 for (int id = 0; id < max_slaves; id++) {
664 pthread_create(&thread, NULL, slave_thread, (void *)(long)id);
667 if (proxy_port) {
668 int proxy_sock = port_listen(proxy_port, max_slaves);
669 for (int id = 0; id < max_slaves; id++) {
670 pthread_create(&thread, NULL, proxy_thread, (void *)(long)proxy_sock);