flan/vendor/agent/flan_agent.c

/* flan_agent — the half of the dev loop that lives in the running program.
 *
 * A redefinition arrives as a path to a .so (runtime/flan_dev.c and
 * Emit.redefinition are what put it there). Two things have to happen to it,
 * and they must happen on different threads:
 *
 *   dlopen   relocates the module and runs the loader. It is milliseconds,
 *            unbounded, and takes the loader lock. Doing it on the game thread
 *            is a dropped frame.
 *   install  is one store per redefined function. It is sub-microsecond, and
 *            it must happen at a point where no redefined function is on the
 *            stack — a frame boundary — or a frame runs half in the old code
 *            and half in the new.
 *
 * So the listener thread does the loading and hands over a function pointer;
 * the game thread calls flan-poll or flan-wait when it is between frames and
 * that is when the swap becomes visible. Nothing else in the program needs to
 * know the agent exists.
 *
 * Nothing is ever dlclosed: a cell holds an address inside a module's text,
 * and unloading it would leave every call site pointing at unmapped memory.
 *
 * This is not a protocol. One line per request, the path to load, and a one
 * line answer. The daemon and its nREPL are a separate program that will speak
 * to a socket, not something this file grows into.
 */

#include <dlfcn.h>
#include <errno.h>
#include <stdlib.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <time.h>
#include <unistd.h>

typedef void (*install_fn)(void);

/* A module may also carry a thunk to run once — that is C-x C-e, an expression
 * compiled into a function with nowhere to be called from. It runs where the
 * install happens, on the game thread between frames, because an expression
 * that reads the program's state has to see it at a point the program agrees
 * is consistent. */
typedef void (*call_fn)(void);

const char *flan_dev_result_get(uint64_t *gen, uint64_t *len);

/* A ring the listener writes and the game thread reads. One producer, one
 * consumer, so two atomics and no lock — the game thread must never block on
 * the loader. Overflow drops the oldest request rather than stalling; a dev
 * loop that queues 64 reloads between two frames has a bigger problem. */
#define QUEUE 64
/* [handle] is set only for a module that declared itself transient — one that
 * ran a thunk and left nothing behind. Everything else is kept mapped forever:
 * a cell holds an address inside a module's text, and unloading it would leave
 * every call site pointing at unmapped memory. */
typedef struct { install_fn install; call_fn call; void *handle; } job;

static job           queue[QUEUE];
static atomic_uint   head;   /* written by the listener */
static atomic_uint   tail;   /* written by the game thread */

static int listen_fd = -1;
static pthread_t listener;
static atomic_int started;

static void publish(job j) {
  unsigned h = atomic_load_explicit(&head, memory_order_relaxed);
  queue[h % QUEUE] = j;
  /* Release: the store to the slot must be visible before the index that
   * advertises it. */
  atomic_store_explicit(&head, h + 1, memory_order_release);
}

/* Returns how many modules were installed. Call it between frames. */
int32_t flan_agent_poll(void) {
  unsigned t = atomic_load_explicit(&tail, memory_order_relaxed);
  unsigned h = atomic_load_explicit(&head, memory_order_acquire);
  int32_t n = 0;
  while (t != h) {
    job j = queue[t % QUEUE];
    t++;
    if (j.install != NULL) { j.install(); n++; }
    /* After the install, so a thunk sees the bodies its own module published. */
    if (j.call != NULL) { j.call(); }
    if (j.handle != NULL) { dlclose(j.handle); }
  }
  atomic_store_explicit(&tail, t, memory_order_relaxed);
  return n;
}

/* The same, but waits up to [ms] for something to arrive first. A game loop
 * does not want this; a headless test does, because it makes the reload
 * deterministic instead of a race against the frame rate. */
int32_t flan_agent_wait(int32_t ms) {
  struct timespec step = { 0, 1000000 };   /* 1ms */
  for (int32_t i = 0; i < ms; i++) {
    int32_t n = flan_agent_poll();
    if (n > 0) return n;
    nanosleep(&step, NULL);
  }
  return flan_agent_poll();
}

/* MSG_NOSIGNAL rather than write(2). A reply goes out in more than one piece,
 * and a sender that has read enough and closed leaves the rest of it writing
 * into a closed socket — which is SIGPIPE, whose default action would kill the
 * program the agent is embedded in. Suppressing it per call rather than
 * installing a handler, because the disposition belongs to the program and not
 * to us. */
static void reply(int fd, const char *s) {
  size_t n = strlen(s);
  while (n > 0) {
    ssize_t k = send(fd, s, n, MSG_NOSIGNAL);
    if (k <= 0) return;
    s += k;
    n -= (size_t)k;
  }
}

/* One connection, one line, one module. Loading here rather than in the game
 * thread is the whole reason this thread exists. */
static void serve(int fd) {
  char line[4096];
  size_t n = 0;
  for (;;) {
    ssize_t k = read(fd, line + n, sizeof line - n - 1);
    if (k <= 0) return;
    n += (size_t)k;
    line[n] = '\0';
    char *nl = strchr(line, '\n');
    if (nl == NULL) {
      if (n == sizeof line - 1) { reply(fd, "err path too long\n"); return; }
      continue;
    }
    *nl = '\0';
    /* The one verb that is not a module: read back the value of the last
     * expression evaluated, with the counter that says whether it is a new
     * one. The daemon polls this rather than the agent holding a connection
     * open across a frame boundary it does not control. */
    if (strcmp(line, "result") == 0) {
      uint64_t gen = 0, len = 0;
      const char *v = flan_dev_result_get(&gen, &len);
      char hdr[64];
      int k = snprintf(hdr, sizeof hdr, "%llu %llu\n", (unsigned long long)gen,
                       (unsigned long long)len);
      if (k > 0) {
        send(fd, hdr, (size_t)k, MSG_NOSIGNAL);
        if (len > 0) send(fd, v, (size_t)len, MSG_NOSIGNAL);
      }
      return;
    }
    void *h = dlopen(line, RTLD_NOW | RTLD_LOCAL);
    if (h == NULL) {
      reply(fd, "err ");
      reply(fd, dlerror());
      reply(fd, "\n");
      return;
    }
    install_fn f = (install_fn)(uintptr_t)dlsym(h, "flan_reload_install");
    if (f == NULL) { reply(fd, "err no flan_reload_install\n"); return; }
    /* Optional: only an expression evaluation has one. */
    call_fn c = (call_fn)(uintptr_t)dlsym(h, "flan_reload_call");
    /* And only one that leaves nothing behind may be unloaded. */
    void *transient =
      (c == NULL) ? NULL : dlsym(h, "flan_reload_transient");
    /* Answer before queueing, not after. The game thread can install and run
     * to completion between the two, and a program that exits there would tear
     * down this connection with the reply still unwritten — which reaches the
     * sender as a reset, not as an answer. */
    reply(fd, "ok\n");
    /* "queued", not "installed": the store happens on the game thread, at a
     * time this thread does not get to choose. */
    publish((job){ f, c, transient == NULL ? NULL : h });
    return;
  }
}

static void *accept_loop(void *arg) {
  (void)arg;
  for (;;) {
    int fd = accept(listen_fd, NULL, NULL);
    if (fd < 0) { if (errno == EINTR) continue; return NULL; }
    serve(fd);
    close(fd);
  }
}

/* [path] is a Flan string: ptr and len, not NUL-terminated.
 *
 * FLAN_AGENT_SOCKET overrides it. A program's source has to name some path,
 * and the daemon that launches the program is the one that knows where it
 * wants to talk to it — without the override the daemon would have to guess,
 * and guessing wrong fails silently: everything compiles, the module is built,
 * and nothing ever receives it. */
int32_t flan_agent_start(const uint8_t *path, int64_t len) {
  struct sockaddr_un addr;
  const char *env = getenv("FLAN_AGENT_SOCKET");
  if (atomic_exchange(&started, 1)) return 0;
  if (env != NULL && env[0] != '\0') {
    path = (const uint8_t *)env;
    len = (int64_t)strlen(env);
  }
  if (len <= 0 || (size_t)len >= sizeof addr.sun_path) return -1;
  memset(&addr, 0, sizeof addr);
  addr.sun_family = AF_UNIX;
  memcpy(addr.sun_path, path, (size_t)len);
  unlink(addr.sun_path);
  listen_fd = socket(AF_UNIX, SOCK_STREAM, 0);
  if (listen_fd < 0) return -1;
  if (bind(listen_fd, (struct sockaddr *)&addr, sizeof addr) < 0) return -1;
  if (listen(listen_fd, 4) < 0) return -1;
  if (pthread_create(&listener, NULL, accept_loop, NULL) != 0) return -1;
  return 0;
}