Distributed engine: Define hook to get stats received from slaves

[pachi/json.git] / distributed / protocol.c

/* The functions implementing the master-slave protocol of the
 * distributed engine are grouped here. They are independent
 * of the gtp protocol. See the comments at the top of distributed.c
 * for a general introduction to the distributed engine. */

#include <assert.h>
#include <stdio.h>
#include <pthread.h>
#include <ctype.h>

#define DEBUG

#include "random.h"
#include "timeinfo.h"
#include "playout.h"
#include "network.h"
#include "debug.h"
#include "distributed/distributed.h"
#include "distributed/protocol.h"

/* All gtp commands for current game separated by \n */
static char gtp_cmds[CMDS_SIZE];

/* Latest gtp command sent to slaves. */
static char *gtp_cmd = NULL;

/* Slaves send gtp_cmd when cmd_count changes. */
static int cmd_count = 0;

/* Remember at most 10 gtp ids per move: kgs-rules, boardsize, clear_board,
 * time_settings, komi, handicap, genmoves, play pass, play pass, final_status_list */
#define MAX_CMDS_PER_MOVE 10

/* History of gtp commands sent for current game, indexed by move. */
static struct cmd_history {
        int gtp_id;
        char *next_cmd;
} history[MAX_GAMELEN][MAX_CMDS_PER_MOVE];

/* Number of active slave machines working for this master. */
int active_slaves = 0;

/* Number of replies to last gtp command already received. */
int reply_count = 0;

/* All replies to latest gtp command are in gtp_replies[0..reply_count-1]. */
char **gtp_replies;


struct receive_buf *receive_queue;
int queue_length = 0;
static int queue_max_length;

/* Mutex protecting all variables above. receive_queue may be
 * read without the lock but is only written with lock held. */
static pthread_mutex_t slave_lock = PTHREAD_MUTEX_INITIALIZER;

/* Condition signaled when a new gtp command is available. */
static pthread_cond_t cmd_cond = PTHREAD_COND_INITIALIZER;

/* Condition signaled when reply_count increases. */
static pthread_cond_t reply_cond = PTHREAD_COND_INITIALIZER;

/* Mutex protecting stderr. Must not be held at same time as slave_lock. */
static pthread_mutex_t log_lock = PTHREAD_MUTEX_INITIALIZER;

/* Absolute time when this program was started.
 * For debugging only. */
static double start_time;

/* Default slave state. */
struct slave_state default_sstate;


/* Get exclusive access to the threads and commands state. */
void
protocol_lock(void)
{
        pthread_mutex_lock(&slave_lock);
}

/* Release exclusive access to the threads and commands state. */
void
protocol_unlock(void)
{
        pthread_mutex_unlock(&slave_lock);
}

/* Write the time, client address, prefix, and string s to stderr atomically.
 * s should end with a \n */
void
logline(struct in_addr *client, char *prefix, char *s)
{
        double now = time_now();

        char addr[INET_ADDRSTRLEN];
        if (client) {
                inet_ntop(AF_INET, client, addr, sizeof(addr));
        } else {
                addr[0] = '\0';
        }
        pthread_mutex_lock(&log_lock);
        fprintf(stderr, "%s%15s %9.3f: %s", prefix, addr, now - start_time, s);
        pthread_mutex_unlock(&log_lock);
}

/* Thread opening a connection on the given socket and copying input
 * from there to stderr. */
static void *
proxy_thread(void *arg)
{
        int proxy_sock = (long)arg;
        assert(proxy_sock >= 0);
        for (;;) {
                struct in_addr client;
                int conn = open_server_connection(proxy_sock, &client);
                FILE *f = fdopen(conn, "r");
                char buf[BSIZE];
                while (fgets(buf, BSIZE, f)) {
                        logline(&client, "< ", buf);
                }
                fclose(f);
        }
}

/* Get a reply to one gtp command. Return the gtp command id,
 * or -1 if error. reply must have at least CMDS_SIZE bytes.
 * The ascii reply ends with an empty line; if the first line
 * contains "@size", a binary reply of size bytes follows the
 * empty line. @size is not standard gtp, it is only used
 * internally by Pachi for the genmoves command; it must be the
 * last parameter on the line.
 * *bin_size is the maximum size upon entry, actual size on return.
 * slave_lock is not held on either entry or exit of this function. */
static int
get_reply(FILE *f, struct in_addr client, char *reply, void *bin_reply, int *bin_size)
{
        int reply_id = -1;
        *reply = '\0';
        if (!fgets(reply, CMDS_SIZE, f)) return -1;

        /* Check for binary reply. */
        char *s = strchr(reply, '@');
        int size = 0;
        if (s) size = atoi(s+1);
        assert(size <= *bin_size);
        *bin_size = size;

        if (DEBUGV(s, 2))
                logline(&client, "<<", reply);
        if ((*reply == '=' || *reply == '?') && isdigit(reply[1]))
                reply_id = atoi(reply+1);

        /* Read the rest of the ascii reply */
        char *line = reply + strlen(reply);
        while (fgets(line, reply + CMDS_SIZE - line, f) && *line != '\n') {
                if (DEBUGL(3))
                        logline(&client, "<<", line);
                line += strlen(line);
        }
        if (*line != '\n') return -1;

        /* Read the binary reply if any. */
        double start = time_now();
        int len;
        while (size && (len = fread(bin_reply, 1, size, f)) > 0) {
                bin_reply = (char *)bin_reply + len;
                size -= len;
        }
        if (*bin_size && DEBUGVV(2)) {
                char buf[1024];
                snprintf(buf, sizeof(buf), "read reply %d bytes in %.4fms\n", *bin_size,
                         (time_now() - start)*1000);
                logline(&client, "= ", buf);
        }
        return size ? -1 : reply_id;
}

/* Send the gtp command to_send and get a reply from the slave machine.
 * Write the reply in buf which must have at least CMDS_SIZE bytes.
 * If *bin_size > 0, send bin_buf after the gtp command.
 * Return any binary reply in bin_buf and set its size in bin_size.
 * bin_buf is private to the slave and need not be copied.
 * Return the gtp command id, or -1 if error.
 * slave_lock is held on both entry and exit of this function. */
static int
send_command(char *to_send, void *bin_buf, int *bin_size,
             FILE *f, struct slave_state *sstate, char *buf)
{
        assert(to_send && gtp_cmd && bin_buf && bin_size);
        strncpy(buf, to_send, CMDS_SIZE);
        bool resend = to_send != gtp_cmd;

        pthread_mutex_unlock(&slave_lock);

        if (DEBUGL(1) && resend)
                logline(&sstate->client, "? ",
                        to_send == gtp_cmds ? "resend all\n" : "partial resend\n");
        fputs(buf, f);

        double start = time_now();
        if (*bin_size)
                fwrite(bin_buf, 1, *bin_size, f);
        fflush(f);

        if (DEBUGV(strchr(buf, '@'), 2)) {
                double ms = (time_now() - start) * 1000.0;
                if (!DEBUGL(3)) {
                        char *s = strchr(buf, '\n');
                        if (s) s[1] = '\0';
                }
                logline(&sstate->client, ">>", buf);
                if (*bin_size) {
                        char b[1024];
                        snprintf(b, sizeof(b),
                                 "sent args %d bytes in %.4fms\n", *bin_size, ms);
                        logline(&sstate->client, "= ", b);
                }
        }

        /* Reuse the buffers for the reply. */
        *bin_size = sstate->max_buf_size;
        int reply_id = get_reply(f, sstate->client, buf, bin_buf, bin_size);

        pthread_mutex_lock(&slave_lock);
        return reply_id;
}

/* Return the command sent after that with the given gtp id,
 * or gtp_cmds if the id wasn't used in this game. If a play command
 * has overwritten a genmoves command, return the play command.
 * slave_lock is held on both entry and exit of this function. */
static char *
next_command(int cmd_id)
{
        if (cmd_id == -1) return gtp_cmds;

        int last_id = atoi(gtp_cmd);
        int reply_move = move_number(cmd_id);
        if (reply_move > move_number(last_id)) return gtp_cmds;

        int slot;
        for (slot = 0; slot < MAX_CMDS_PER_MOVE; slot++) {
                if (cmd_id == history[reply_move][slot].gtp_id) break;
        }
        if (slot == MAX_CMDS_PER_MOVE) return gtp_cmds;

        char *next = history[reply_move][slot].next_cmd;
        assert(next);
        return next;
}

/* Allocate buffers for a slave thread. The state should have been
 * initialized already as a copy of the default slave state.
 * slave_lock is not held on either entry or exit of this function. */
static void
slave_state_alloc(struct slave_state *sstate)
{
        for (int n = 0; n < BUFFERS_PER_SLAVE; n++) {
                sstate->b[n].buf = malloc2(sstate->max_buf_size);
        }
        if (sstate->alloc_hook) sstate->alloc_hook(sstate);
}

/* Get a free binary buffer, first invalidating it in the receive
 * queue if necessary. In practice all buffers should be used
 * before they are invalidated, if BUFFERS_PER_SLAVE is large enough.
 * slave_lock is held on both entry and exit of this function. */
static void *
get_free_buf(struct slave_state *sstate, bool new_id)
{
        int newest = (sstate->newest_buf + 1) & (BUFFERS_PER_SLAVE - 1);
        sstate->newest_buf = newest;
        void *buf = sstate->b[newest].buf;

        if (DEBUGVV(7)) {
                char b[1024];
                snprintf(b, sizeof(b), "get free %d index %d buf=%p qlength %d\n",
                         newest, sstate->b[newest].queue_index, buf, queue_length);
                logline(&sstate->client, "? ", b);
        }

        /* For a new command, previous indices in receive_queue
         * are now meaningless. In particular they may be
         * beyond the current queue_length. */
        if (new_id) {
                sstate->last_processed = -1;
                for (int n = 0; n < BUFFERS_PER_SLAVE; n++) {
                        sstate->b[n].queue_index = -1;
                }
                return buf;
        }

        int index = sstate->b[newest].queue_index;
        if (index >= 0) {
                assert(receive_queue[index].thread_id == sstate->thread_id);
                assert(receive_queue[index].buf == buf);
                /* Invalidate the buffer. */
                receive_queue[index].buf = NULL;
                sstate->b[newest].queue_index = -1;
        }
        return buf;
}

/* Insert a buffer in the receive queue. It should be the most
 * recent buffer allocated by the calling thread.
 * slave_lock is held on both entry and exit of this function. */
static void
insert_buf(struct slave_state *sstate, void *buf, int size)
{
        assert(queue_length < queue_max_length);

        int newest = sstate->newest_buf;
        assert(buf == sstate->b[newest].buf);

        /* Update the buffer if necessary before making it
         * available to other threads. */
        if (sstate->insert_hook) sstate->insert_hook(buf, size);

        if (DEBUGVV(7)) {
                char b[1024];
                snprintf(b, sizeof(b), "insert newest %d rq[%d]->%p\n",
                         newest, queue_length, buf);
                        logline(&sstate->client, "? ", b);
        }
        receive_queue[queue_length].buf = buf;
        receive_queue[queue_length].size = size;
        receive_queue[queue_length].thread_id = sstate->thread_id;
        sstate->b[newest].queue_index = queue_length;
        queue_length++;
}

/* Process the reply received from a slave machine.
 * Copy the ascii part to reply_buf and insert the binary part
 * (if any) in the receive queue.
 * Return false if ok, true if the slave is out of sync.
 * slave_lock is held on both entry and exit of this function. */
static bool
process_reply(int reply_id, char *reply, char *reply_buf,
              void *bin_reply, int bin_size, int *last_reply_id,
              int *reply_slot, struct slave_state *sstate)
{
        /* Resend everything if slave returned an error. */
        if (*reply != '=') {
                *last_reply_id = -1;
                return true;
        }
        /* Make sure we are still in sync. cmd_count may have
         * changed but the reply is valid as long as cmd_id didn't
         * change (this only occurs for consecutive genmoves). */
        int cmd_id = atoi(gtp_cmd);
        if (reply_id != cmd_id) {
                *last_reply_id = reply_id;
                return true;
        }

        strncpy(reply_buf, reply, CMDS_SIZE);
        if (reply_id != *last_reply_id)
                *reply_slot = reply_count++;
        gtp_replies[*reply_slot] = reply_buf;

        if (bin_size) insert_buf(sstate, bin_reply, bin_size);

        pthread_cond_signal(&reply_cond);
        *last_reply_id = reply_id;
        return false;
}

/* Get the binary arg for the given command, and update the command
 * if necessary. For now, only genmoves has a binary argument, and
 * we return the best stats increments from all other slaves.
 * Set *bin_size to 0 if the command doesn't take binary arguments,
 * but still return a buffer, to be used for the reply.
 * Return NULL if the binary arg is obsolete by the time we have
 * finished computing it, because a new command is available.
 * This version only gets the buffer for the reply, to be completed
 * in future commits.
 * slave_lock is held on both entry and exit of this function. */
void *
get_binary_arg(struct slave_state *sstate, char *cmd, int cmd_size, int *bin_size)
{
        int cmd_id = atoi(gtp_cmd);
        void *buf = get_free_buf(sstate, cmd_id != sstate->last_cmd_id);
        sstate->last_cmd_id = cmd_id;

        *bin_size = 0;
        char *s = strchr(cmd, '@');
        if (!s || !sstate->args_hook) return buf;

        int size = sstate->args_hook(buf, sstate, cmd_id);

        /* Check that the command is still valid. */
        if (atoi(gtp_cmd) != cmd_id) return NULL;

        /* Set the correct binary size for this slave.
         * cmd may have been overwritten with new parameters. */
        *bin_size = size;
        s = strchr(cmd, '@');
        assert(s);
        snprintf(s, cmd + cmd_size - s, "@%d\n", size);
        return buf;
}

/* Main loop of a slave thread.
 * Send the current command to the slave machine and wait for a reply.
 * Resend command history if the slave machine is out of sync.
 * Returns when the connection with the slave machine is cut.
 * slave_lock is held on both entry and exit of this function. */
static void
slave_loop(FILE *f, char *reply_buf, struct slave_state *sstate, bool resend)
{
        char *to_send;
        int last_cmd_count = 0;
        int last_reply_id = -1;
        int reply_slot = -1;
        for (;;) {
                if (resend) {
                        /* Resend complete or partial history */
                        to_send = next_command(last_reply_id);
                } else {
                        /* Wait for a new command. */
                        while (last_cmd_count == cmd_count)
                                pthread_cond_wait(&cmd_cond, &slave_lock);
                        to_send = gtp_cmd;
                }

                /* Command available, send it to slave machine.
                 * If slave was out of sync, send the history.
                 * But first get binary arguments if necessary. */
                int bin_size = 0;
                void *bin_buf = get_binary_arg(sstate, gtp_cmd,
                                               gtp_cmds + CMDS_SIZE - gtp_cmd,
                                               &bin_size);
                /* Check that the command is still valid. */
                resend = true;
                if (!bin_buf) continue;

                /* Send the command and get the reply, which always ends with \n\n
                 * The slave machine sends "=id reply" or "?id reply"
                 * with id == cmd_id if it is in sync. */
                last_cmd_count = cmd_count;
                char buf[CMDS_SIZE];
                int reply_id = send_command(to_send, bin_buf, &bin_size, f,
                                            sstate, buf);
                if (reply_id == -1) return;

                resend = process_reply(reply_id, buf, reply_buf, bin_buf, bin_size,
                                       &last_reply_id, &reply_slot, sstate);
        }
}

/* Thread sending gtp commands to one slave machine, and
 * reading replies. If a slave machine dies, this thread waits
 * for a connection from another slave.
 * The large buffers are allocated only once we get a first
 * connection, to avoid wasting memory if max_slaves is too large.
 * We do not invalidate the received buffers if a slave disconnects;
 * they are still useful for other slaves. */
static void *
slave_thread(void *arg)
{
        struct slave_state sstate = default_sstate;
        sstate.thread_id = (long)arg;

        assert(sstate.slave_sock >= 0);
        char reply_buf[CMDS_SIZE];
        bool resend = false;

        for (;;) {
                /* Wait for a connection from any slave. */
                struct in_addr client;
                int conn = open_server_connection(sstate.slave_sock, &client);

                FILE *f = fdopen(conn, "r+");
                if (DEBUGL(2)) {
                        snprintf(reply_buf, sizeof(reply_buf),
                                 "new slave, id %d\n", sstate.thread_id);
                        logline(&client, "= ", reply_buf);
                }

                /* Minimal check of the slave identity. */
                fputs("name\n", f);
                if (!fgets(reply_buf, sizeof(reply_buf), f)
                    || strncasecmp(reply_buf, "= Pachi", 7)
                    || !fgets(reply_buf, sizeof(reply_buf), f)
                    || strcmp(reply_buf, "\n")) {
                        logline(&client, "? ", "bad slave\n");
                        fclose(f);
                        continue;
                }

                if (!resend) slave_state_alloc(&sstate);
                sstate.client = client;

                pthread_mutex_lock(&slave_lock);
                active_slaves++;
                slave_loop(f, reply_buf, &sstate, resend);

                assert(active_slaves > 0);
                active_slaves--;
                // Unblock main thread if it was waiting for this slave.
                pthread_cond_signal(&reply_cond);
                pthread_mutex_unlock(&slave_lock);

                resend = true;
                if (DEBUGL(2))
                        logline(&client, "= ", "lost slave\n");
                fclose(f);
        }
}

/* Create a new gtp command for all slaves. The slave lock is held
 * upon entry and upon return, so the command will actually be
 * sent when the lock is released. The last command is overwritten
 * if gtp_cmd points to a non-empty string. cmd is a single word;
 * args has all arguments and is empty or has a trailing \n */
void
update_cmd(struct board *b, char *cmd, char *args, bool new_id)
{
        assert(gtp_cmd);
        /* To make sure the slaves are in sync, we ignore the original id
         * and use the board number plus some random bits as gtp id. */
        static int gtp_id = -1;
        int moves = is_reset(cmd) ? 0 : b->moves;
        if (new_id) {
                int prev_id = gtp_id;
                do {
                        /* fast_random() is 16-bit only so the multiplication can't overflow. */
                        gtp_id = force_reply(moves + fast_random(65535) * DIST_GAMELEN);
                } while (gtp_id == prev_id);
                reply_count = 0;
        }
        snprintf(gtp_cmd, gtp_cmds + CMDS_SIZE - gtp_cmd, "%d %s %s",
                 gtp_id, cmd, *args ? args : "\n");
        cmd_count++;

        /* Remember history for out-of-sync slaves. */
        static int slot = 0;
        static struct cmd_history *last = NULL;
        if (new_id) {
                if (last) last->next_cmd = gtp_cmd;
                slot = (slot + 1) % MAX_CMDS_PER_MOVE;
                last = &history[moves][slot];
                last->gtp_id = gtp_id;
                last->next_cmd = NULL;
        }
        // Notify the slave threads about the new command.
        pthread_cond_broadcast(&cmd_cond);
}

/* Update the command history, then create a new gtp command
 * for all slaves. The slave lock is held upon entry and
 * upon return, so the command will actually be sent when the
 * lock is released. cmd is a single word; args has all
 * arguments and is empty or has a trailing \n */
void
new_cmd(struct board *b, char *cmd, char *args)
{
        // Clear the history when a new game starts:
        if (!gtp_cmd || is_gamestart(cmd)) {
                gtp_cmd = gtp_cmds;
                memset(history, 0, sizeof(history));
        } else {
                /* Preserve command history for new slaves.
                 * To indicate that the slave should only reply to
                 * the last command we force the id of previous
                 * commands to be just the move number. */
                int id = prevent_reply(atoi(gtp_cmd));
                int len = strspn(gtp_cmd, "0123456789");
                char buf[32];
                snprintf(buf, sizeof(buf), "%0*d", len, id);
                memcpy(gtp_cmd, buf, len);

                gtp_cmd += strlen(gtp_cmd);
        }

        // Let the slave threads send the new gtp command:
        update_cmd(b, cmd, args, true);
}

/* Wait for at least one new reply. Return when at least
 * min_replies slaves have already replied, or when the
 * given absolute time is passed.
 * The replies are returned in gtp_replies[0..reply_count-1]
 * slave_lock is held on entry and on return. */
void
get_replies(double time_limit, int min_replies)
{
        for (;;) {
                if (reply_count > 0) {
                        struct timespec ts;
                        double sec;
                        ts.tv_nsec = (int)(modf(time_limit, &sec)*1000000000.0);
                        ts.tv_sec = (int)sec;
                        pthread_cond_timedwait(&reply_cond, &slave_lock, &ts);
                } else {
                        pthread_cond_wait(&reply_cond, &slave_lock);
                }
                if (reply_count == 0) continue;
                if (reply_count >= min_replies || reply_count >= active_slaves) return;
                if (time_now() >= time_limit) break;
        }
        if (DEBUGL(1)) {
                char buf[1024];
                snprintf(buf, sizeof(buf),
                         "get_replies timeout %.3f >= %.3f, replies %d < min %d, active %d\n",
                         time_now() - start_time, time_limit - start_time,
                         reply_count, min_replies, active_slaves);
                logline(NULL, "? ", buf);
        }
        assert(reply_count > 0);
}

/* In a 30s move with at least 5ms per genmoves we get at most
 * 6000 genmoves per slave. */
#define MAX_GENMOVES_PER_SLAVE 6000

/* Allocate the receive queue, and create the slave and proxy threads.
 * max_buf_size and the merge-related fields of default_sstate must
 * already be initialized. */
void
protocol_init(char *slave_port, char *proxy_port, int max_slaves)
{
        start_time = time_now();

        queue_max_length = max_slaves * MAX_GENMOVES_PER_SLAVE;
        receive_queue = calloc2(queue_max_length, sizeof(*receive_queue));

        default_sstate.slave_sock = port_listen(slave_port, max_slaves);
        default_sstate.last_processed = -1;

        for (int n = 0; n < BUFFERS_PER_SLAVE; n++) {
                default_sstate.b[n].queue_index = -1;
        }

        pthread_t thread;
        for (int id = 0; id < max_slaves; id++) {
                pthread_create(&thread, NULL, slave_thread, (void *)(long)id);
        }

        if (proxy_port) {
                int proxy_sock = port_listen(proxy_port, max_slaves);
                for (int id = 0; id < max_slaves; id++) {
                        pthread_create(&thread, NULL, proxy_thread, (void *)(long)proxy_sock);
                }
        }
}