text: /**
   * \file
   * \brief Process management service.
   */
  
  /*
   * Copyright (c) 2017, ETH Zurich.
   * All rights reserved.
   *
   * This file is distributed under the terms in the attached LICENSE file.
   * If you do not find this file, copies can be found by writing to:
   * ETH Zurich D-INFK, Universitaetstrasse 6, CH-8092 Zurich. Attn: Systems Group.
   */
  
  #include <barrelfish/barrelfish.h>
  #include <barrelfish/nameservice_client.h>
  #include <barrelfish/spawn_client.h>
  #include <if/monitor_defs.h>
  #include <if/proc_mgmt_defs.h>
  #include <if/spawn_defs.h>
  
  #include <dist/barrier.h>
  
  #include "domain.h"
  #include "internal.h"
  #include "pending_clients.h"
  #include "spawnd_state.h"
  
  /**
   * \brief Handler for message add_spawnd, for the local monitor binding.
   */
  static void add_spawnd_handler(struct proc_mgmt_binding *b, coreid_t core_id,
                                 iref_t iref)
  {
      if (spawnd_state_exists(core_id)) {
          DEBUG_ERR(PROC_MGMT_ERR_SPAWND_EXISTS, "spawnd_state_exists");
          return;
      }
  
      // Bind with the spawnd.
      struct spawn_binding *spawnb;
      errval_t err = spawn_bind_iref(iref, &spawnb);
      if (err_is_fail(err)) {
          DEBUG_ERR(err, "spawn_bind_iref");
          return;
      }
  
      err = spawnd_state_alloc(core_id, spawnb);
      if (err_is_fail(err)) {
          DEBUG_ERR(err, "spawnd_state_alloc");
      }
  
      // Tell name service that we're ready on the core spawnd just came up
      err = nsb_register_n(core_id, SERVICE_BASENAME);
      if (err_is_fail(err)) {
          USER_PANIC_ERR(err, "nsb_register_n failed");
      }
  }
  
  /**
   * \brief Handler for message add_spawnd, for non-monitor bindings.
   */
  static void add_spawnd_handler_non_monitor(struct proc_mgmt_binding *b,
                                             coreid_t core_id, iref_t iref)
  {
      // debug_printf("Ignoring add_spawnd call: %s\n",
      //              err_getstring(PROC_MGMT_ERR_NOT_MONITOR));
  }
  
  static bool cleanup_request_sender(struct msg_queue_elem *m);
  
  /**
   * General-purpose handler for replies from spawnd.
   */
  static void spawn_reply_handler(struct spawn_binding *b, errval_t spawn_err)
  {
      struct pending_client *cl =
              (struct pending_client*) spawnd_state_dequeue_recv(b->st);
  
      /* there's no pending client, return */
      if (cl == NULL) {
          return;    
      }
  
      struct pending_spawn *spawn = NULL;
      struct pending_span *span = NULL;
      struct pending_kill_cleanup *kc = NULL;
  
      struct domain_entry *entry;
      
      errval_t err, resp_err;
  
      switch (cl->type) {
          case ClientType_Spawn:
          case ClientType_SpawnWithCaps:
              spawn = (struct pending_spawn*) cl->st;
              err = spawn_err;
              if (err_is_ok(spawn_err)) {
                  err = domain_spawn(spawn->cap_node, spawn->core_id, spawn->argvbuf,
                                     spawn->argvbytes);
              }
              if (cl->type == ClientType_Spawn) {
                  resp_err = cl->b->tx_vtbl.spawn_response(cl->b, NOP_CONT,
                          err, spawn->cap_node->domain_cap);
              } else {
                  resp_err = cl->b->tx_vtbl.spawn_with_caps_response(cl->b,
                          NOP_CONT, err, spawn->cap_node->domain_cap);
              }
  
              free(spawn);
              break;
  
          case ClientType_Span:
              span = (struct pending_span*) cl->st;
              entry = span->entry;
              if (entry->status == DOMAIN_STATUS_RUNNING) {
                  resp_err = cl->b->tx_vtbl.span_response(cl->b, NOP_CONT,
                                                          spawn_err);
              }
  
              free(span);
              break;
  
          case ClientType_Cleanup:
              kc = (struct pending_kill_cleanup*) cl->st;
              entry = kc->entry;
  
              assert(entry->num_spawnds_resources > 0);
              assert(entry->status != DOMAIN_STATUS_CLEANED);
  
              --entry->num_spawnds_resources;
              if (entry->num_spawnds_resources == 0) {
                  entry->status = DOMAIN_STATUS_CLEANED;
  
                  // At this point, the domain exists in state CLEANED for history
                  // reasons. For instance, if some other domain issues a wait
                  // call for this one, the process manager can return the exit
                  // status directly. At some point, however, we might want to
                  // just clean up the domain entry and recycle the domain cap.
              }
  
              free(kc);
              break;
  
          case ClientType_Kill:
          case ClientType_Exit:
              kc = (struct pending_kill_cleanup*) cl->st;
              entry = kc->entry;
  
              assert(entry->num_spawnds_running > 0);
              assert(entry->status != DOMAIN_STATUS_STOPPED);
  
              --entry->num_spawnds_running;
  
              if (entry->num_spawnds_running == 0) {
                  entry->status = DOMAIN_STATUS_STOPPED;
  
                  if (cl->type == ClientType_Kill) {
                      entry->exit_status = EXIT_STATUS_KILLED;
                      resp_err = cl->b->tx_vtbl.kill_response(cl->b, NOP_CONT,
                                                              spawn_err);
                  }
  
                  struct domain_waiter *waiter = entry->waiters;
                  while (waiter != NULL) {
                      waiter->b->tx_vtbl.wait_response(waiter->b, NOP_CONT,
                                                       SYS_ERR_OK,
                                                       entry->exit_status);
                      struct domain_waiter *tmp = waiter;
                      waiter = waiter->next;
                      free(tmp);
                  }
  
                  for (coreid_t i = 0; i < MAX_COREID; ++i) {
                      if (entry->spawnds[i] == NULL) {
                          continue;
                      }
  
                      struct spawn_binding *spb = entry->spawnds[i]->b;
  
                      struct pending_kill_cleanup *cleanup =
                              (struct pending_kill_cleanup*) malloc(
                                      sizeof(struct pending_kill_cleanup));
                      cleanup->b = spb;
                      cleanup->domain_cap = kc->domain_cap;
                      cleanup->entry = entry;
  
                      struct pending_client *cleanup_cl =
                              (struct pending_client*) malloc(
                                      sizeof(struct pending_client));
                      cleanup_cl->b = cl->b;
                      cleanup_cl->type = ClientType_Cleanup;
                      cleanup_cl->st = cleanup;
  
                      struct msg_queue_elem *msg = (struct msg_queue_elem*) malloc(
                              sizeof(struct msg_queue_elem));
                      msg->st = cleanup_cl;
                      msg->cont = cleanup_request_sender;
  
                      err = spawnd_state_enqueue_send(entry->spawnds[i], msg);
  
                      if (err_is_fail(err)) {
                          DEBUG_ERR(err, "enqueuing cleanup request");
                          free(cleanup);
                          free(cleanup_cl);
                          free(msg);
                      }
                  }
              }
  
              free(kc);
              break;
  
          default:
              USER_PANIC("Unknown client type in spawn_reply_handler: %u\n",
                         cl->type);
      }
  
      free(cl);
  }
  
  /**
   * \brief Handler for sending spawn requests.
   */
  static bool spawn_request_sender(struct msg_queue_elem *m)
  {
      struct pending_client *cl = (struct pending_client*) m->st;
      struct pending_spawn *spawn = (struct pending_spawn*) cl->st;
      spawn->b->rx_vtbl.spawn_reply = spawn_reply_handler;
  
      errval_t err;
      bool with_caps = !(capref_is_null(spawn->inheritcn_cap) &&
                         capref_is_null(spawn->argcn_cap));
      if (with_caps) {
          err = spawn->b->tx_vtbl.spawn_with_caps_request(spawn->b, NOP_CONT,
                                                          cap_procmng,
                                                          spawn->cap_node->domain_cap,
                                                          spawn->path,
                                                          spawn->argvbuf,
                                                          spawn->argvbytes,
                                                          spawn->envbuf,
                                                          spawn->envbytes,
                                                          spawn->inheritcn_cap,
                                                          spawn->argcn_cap,
                                                          spawn->flags);
      } else {
          err = spawn->b->tx_vtbl.spawn_request(spawn->b, NOP_CONT, cap_procmng,
                                                spawn->cap_node->domain_cap,
                                                spawn->path, spawn->argvbuf,
                                                spawn->argvbytes, spawn->envbuf,
                                                spawn->envbytes, spawn->flags);
      }
  
      if (err_is_fail(err)) {
          if (err_no(err) == FLOUNDER_ERR_TX_BUSY) {
              return false;
          } else {
              USER_PANIC_ERR(err, "sending spawn request");
          }
      }
  
      free(m);
  
      return true;
  }
  
  /**
   * \brief Handler for sending span requests.
   */
  static bool span_request_sender(struct msg_queue_elem *m)
  {
      struct pending_client *cl = (struct pending_client*) m->st;
      struct pending_span *span = (struct pending_span*) cl->st;
  
      errval_t err;
      span->b->rx_vtbl.spawn_reply = spawn_reply_handler;
      err = span->b->tx_vtbl.span_request(span->b, NOP_CONT, cap_procmng,
                                          span->domain_cap, span->vroot,
                                          span->dispframe);
  
      if (err_is_fail(err)) {
          if (err_no(err) == FLOUNDER_ERR_TX_BUSY) {
              return false;
          } else {
              USER_PANIC_ERR(err, "sending span request");
          }
      }
  
      free(m);
  
      return true;
  }
  
  /**
   * \brief Handler for sending kill requests.
   */
  static bool kill_request_sender(struct msg_queue_elem *m)
  {
      struct pending_client *cl = (struct pending_client*) m->st;
      struct pending_kill_cleanup *kill = (struct pending_kill_cleanup*) cl->st;
  
      errval_t err;
      kill->b->rx_vtbl.spawn_reply = spawn_reply_handler;
      err = kill->b->tx_vtbl.kill_request(kill->b, NOP_CONT, cap_procmng,
                                          kill->domain_cap);
  
      if (err_is_fail(err)) {
          if (err_no(err) == FLOUNDER_ERR_TX_BUSY) {
              return false;
          } else {
              USER_PANIC_ERR(err, "sending kill request");
          }
      }
  
      free(m);
  
      return true;
  }
  
  /**
   * \brief Handler for sending cleanup requests.
   */
  static bool cleanup_request_sender(struct msg_queue_elem *m)
  {
      struct pending_client *cl = (struct pending_client*) m->st;
      struct pending_kill_cleanup *cleanup = (struct pending_kill_cleanup*) cl->st;
  
      errval_t err;
      cleanup->b->rx_vtbl.spawn_reply = spawn_reply_handler;
      err = cleanup->b->tx_vtbl.cleanup_request(cleanup->b, NOP_CONT,
                                                cap_procmng,
                                                cleanup->domain_cap);
  
      if (err_is_fail(err)) {
          if (err_no(err) == FLOUNDER_ERR_TX_BUSY) {
              return false;
          } else {
              USER_PANIC_ERR(err, "sending cleanup request");
          }
      }
  
      free(m);
  
      return true;
  }
  
  /**
   * \brief Common bits of the spawn and spawn_with_caps handlers.
   */
  static errval_t spawn_handler_common(struct proc_mgmt_binding *b,
                                       enum ClientType type,
                                       coreid_t core_id, const char *path,
                                       const char *argvbuf, size_t argvbytes,
                                       const char *envbuf, size_t envbytes,
                                       struct capref inheritcn_cap,
                                       struct capref argcn_cap, uint8_t flags)
  {
      if (!spawnd_state_exists(core_id)) {
          /* Spawnd on the requested core is not up
           * Clients can do nsb_wait_n(core_id, "proc_mgmt") to avoid
           * this situation 
           */
          return PROC_MGMT_ERR_INVALID_SPAWND;
      }
  
      struct spawnd_state *spawnd = spawnd_state_get(core_id);
      assert(spawnd != NULL);
      struct spawn_binding *cl = spawnd->b;
      assert(cl != NULL);
  
      errval_t err;
      if (domain_should_refill_caps()) {
          err = domain_prealloc_caps();
          if (err_is_fail(err)) {
              return err_push(err, PROC_MGMT_ERR_CREATE_DOMAIN_CAP);
          }
      }
  
      struct domain_cap_node *cap_node = next_cap_node();
  
      struct pending_spawn *spawn = (struct pending_spawn*) malloc(
              sizeof(struct pending_spawn));
      spawn->cap_node = cap_node;
      // spawn->domain_cap = domain_cap;
      spawn->b = cl;
      spawn->core_id = core_id;
      spawn->path = path;
      spawn->argvbuf = argvbuf;
      spawn->argvbytes = argvbytes;
      spawn->envbuf = envbuf;
      spawn->envbytes = envbytes;
      spawn->inheritcn_cap = inheritcn_cap;
      spawn->argcn_cap = argcn_cap;
      spawn->flags = flags;
  
      struct pending_client *spawn_cl = (struct pending_client*) malloc(
              sizeof(struct pending_client));
      spawn_cl->b = b;
      spawn_cl->type = type;
      spawn_cl->st = spawn;
  
      struct msg_queue_elem *msg = (struct msg_queue_elem*) malloc(
              sizeof(struct msg_queue_elem));
      msg->st = spawn_cl;
      msg->cont = spawn_request_sender;
  
      err = spawnd_state_enqueue_send(spawnd, msg);
      if (err_is_fail(err)) {
          DEBUG_ERR(err, "enqueuing spawn request");
          free(spawn);
          free(spawn_cl);
          free(msg);
      }
  
      return SYS_ERR_OK;
  }
  
  /**
   * \brief Handler for rpc spawn.
   */
  static void spawn_handler(struct proc_mgmt_binding *b, coreid_t core_id,
                            const char *path, const char *argvbuf,
                            size_t argvbytes, const char *envbuf, size_t envbytes,
                            uint8_t flags)
  {
      errval_t err, resp_err;
      err = spawn_handler_common(b, ClientType_Spawn, core_id, path, argvbuf,
                                 argvbytes, envbuf, envbytes, NULL_CAP, NULL_CAP,
                                 flags);
  
      if (err_is_fail(err)) {
          resp_err = b->tx_vtbl.spawn_response(b, NOP_CONT, err, NULL_CAP);
          if (err_is_fail(resp_err)) {
              DEBUG_ERR(resp_err, "failed to send spawn_response");
          }
      }
  }
  
  /**
   * \brief Handler for rpc spawn_with_caps.
   */
  static void spawn_with_caps_handler(struct proc_mgmt_binding *b,
                                      coreid_t core_id, const char *path,
                                      const char *argvbuf, size_t argvbytes,
                                      const char *envbuf, size_t envbytes,
                                      struct capref inheritcn_cap,
                                      struct capref argcn_cap, uint8_t flags)
  {
      errval_t err, resp_err;
      err = spawn_handler_common(b, ClientType_SpawnWithCaps, core_id, path,
                                 argvbuf, argvbytes, envbuf, envbytes,
                                 inheritcn_cap, argcn_cap, flags);
      if (err_is_ok(err)) {
          // Will respond to client when we get the reply from spawnd.
          return;
      }
  
      resp_err = b->tx_vtbl.spawn_with_caps_response(b, NOP_CONT, err,
                                                              NULL_CAP);
      if (err_is_fail(resp_err)) {
          DEBUG_ERR(resp_err, "failed to send spawn_with_caps_response");
      }
  }
  
  /**
   * \brief Handler for rpc span.
   */
  static void span_handler(struct proc_mgmt_binding *b, struct capref domain_cap,
                           coreid_t core_id, struct capref vroot,
                           struct capref dispframe)
  {
      errval_t err, resp_err;
      struct domain_entry *entry = NULL;
      err = domain_get_by_cap(domain_cap, &entry);
      if (err_is_fail(err)) {
          goto respond_with_err;
      }
  
      assert(entry != NULL);
      if (entry->status != DOMAIN_STATUS_RUNNING) {
          err = PROC_MGMT_ERR_DOMAIN_NOT_RUNNING;
          goto respond_with_err;
      }
  
      if (entry->spawnds[core_id] != NULL) {
          // TODO(razvan): Maybe we want to allow the same domain to span multiple
          // dispatchers onto the same core?
          err = PROC_MGMT_ERR_ALREADY_SPANNED;
          goto respond_with_err;
      }
  
      if (!spawnd_state_exists(core_id)) {
          err = PROC_MGMT_ERR_INVALID_SPAWND;
          goto respond_with_err;
      }
  
      struct spawnd_state *spawnd = spawnd_state_get(core_id);
      assert(spawnd != NULL);
      struct spawn_binding *cl = spawnd->b;
      assert(cl != NULL);
  
      struct pending_span *span = (struct pending_span*) malloc(
              sizeof(struct pending_span));
      span->domain_cap = domain_cap;
      span->entry = entry;
      span->b = cl;
      span->core_id = core_id;
      span->vroot = vroot;
      span->dispframe = dispframe;
  
      struct pending_client *span_cl = (struct pending_client*) malloc(
              sizeof(struct pending_client));
      span_cl->b = b;
      span_cl->type = ClientType_Span;
      span_cl->st = span;
  
      struct msg_queue_elem *msg = (struct msg_queue_elem*) malloc(
              sizeof(struct msg_queue_elem));
      msg->st = span_cl;
      msg->cont = span_request_sender;
  
      err = spawnd_state_enqueue_send(spawnd, msg);
  
      if (err_is_fail(err)) {
          DEBUG_ERR(err, "enqueuing span request");
          free(span);
          free(span_cl);
          free(msg);
      }
  
  respond_with_err:
      resp_err = b->tx_vtbl.span_response(b, NOP_CONT, err);
      if (err_is_fail(resp_err)) {
          DEBUG_ERR(resp_err, "failed to send span_response");
      }
  }
  
  /**
   * \brief Common bits of the kill and exit handlers.
   */
  static errval_t kill_handler_common(struct proc_mgmt_binding *b,
                                      struct capref domain_cap,
                                      enum ClientType type,
                                      uint8_t exit_status)
  {
      struct domain_entry *entry;
      errval_t err = domain_get_by_cap(domain_cap, &entry);
      if (err_is_fail(err)) {
          return err;
      }
  
      entry->exit_status = exit_status;
      domain_stop_pending(entry);
  
      for (coreid_t i = 0; i < MAX_COREID; ++i) {
          if (entry->spawnds[i] == NULL) {
              continue;
          }
  
          struct spawn_binding *spb = entry->spawnds[i]->b;
  
          struct pending_kill_cleanup *cmd = (struct pending_kill_cleanup*) malloc(
                  sizeof(struct pending_kill_cleanup));
          cmd->domain_cap = domain_cap;
          cmd->entry = entry;
          cmd->b = spb;
  
          struct pending_client *cl = (struct pending_client*) malloc(
                  sizeof(struct pending_client));
          cl->b = b;
          cl->type = type;
          cl->st = cmd;
  
          struct msg_queue_elem *msg = (struct msg_queue_elem*) malloc(
                  sizeof(struct msg_queue_elem));
          msg->st = cl;
          msg->cont = kill_request_sender;
  
          err = spawnd_state_enqueue_send(entry->spawnds[i], msg);
          if (err_is_fail(err)) {
              DEBUG_ERR(err, "enqueuing kill request");
              free(cmd);
              free(cl);
              free(msg);
          }
      }
  
      return SYS_ERR_OK;
  }
  
  /**
   * \brief Handler for rpc kill.
   */
  static void kill_handler(struct proc_mgmt_binding *b,
                           struct capref victim_domain_cap)
  {
      errval_t err = kill_handler_common(b, victim_domain_cap, ClientType_Kill,
                                         EXIT_STATUS_KILLED);
      if (err_is_fail(err)) {
          errval_t resp_err = b->tx_vtbl.kill_response(b, NOP_CONT, err);
          if (err_is_fail(resp_err)) {
              DEBUG_ERR(resp_err, "failed to send kill_response");
          }
      }
  }
  
  /**
   * \brief Handler for message exit.
   */
  static void exit_handler(struct proc_mgmt_binding *b, struct capref domain_cap,
                           uint8_t exit_status)
  {
      errval_t err = kill_handler_common(b, domain_cap, ClientType_Exit,
                                         exit_status);
      if (err_is_fail(err)) {
          DEBUG_ERR(err, "processing exit_handler for requesting domain, exit "
                    "code %u", exit_status);
      }
      // Error or not, there's no client to respond to anymore.
  }
  
  /**
   * \brief Handler for rpc wait.
   */
  static void wait_handler(struct proc_mgmt_binding *b, struct capref domain_cap, bool nohang)
  {
      errval_t err, resp_err;
      struct domain_entry *entry;
      err = domain_get_by_cap(domain_cap, &entry);
      if (err_is_fail(err)) {
          goto respond;
      }
  
      if (entry->status == DOMAIN_STATUS_STOPPED) {
          // Domain has already been stopped, so just reply with exit status.
          goto respond;
      }
  
      if (nohang) {
          entry->exit_status = -1;
          goto respond;   
      } 
  
      struct domain_waiter *waiter = (struct domain_waiter*) malloc(
              sizeof(struct domain_waiter));
      waiter->b = b;
      waiter->next = entry->waiters;
      entry->waiters = waiter;
      // Will respond when domain is stopped.
      return;
  
  respond:
      resp_err = b->tx_vtbl.wait_response(b, NOP_CONT, err, entry->exit_status);
      if (err_is_fail(resp_err)) {
          DEBUG_ERR(resp_err, "failed to send wait_response");
      }
  }
  
  /**
   * \brief Handler for rpc get_domainlist.
   */
  static void get_domainlist_handler(struct proc_mgmt_binding *b)
  {
      errval_t resp_err;
      size_t len;
      domainid_t* domains;
  
      domain_get_all_ids(&domains, &len);
  
      // 4096 hardcoded limit in flounder interface
      assert(sizeof(domainid_t)/sizeof(uint8_t)*len < 4096);
  
      resp_err = b->tx_vtbl.get_domainlist_response(b, NOP_CONT, (uint8_t*) domains, 
                                                    sizeof(domainid_t)/sizeof(uint8_t)*len);
      if (err_is_fail(resp_err)) {
          DEBUG_ERR(resp_err, "failed to send wait_response");
      }
  }
  
  /**
   * \brief Handler for rpc get_status.
   */
  static void get_status_handler(struct proc_mgmt_binding *b, domainid_t domain)
  {
      errval_t err;
      struct domain_entry* entry;
      proc_mgmt_ps_entry_t pse;
      memset(&pse, 0, sizeof(pse));
  
      err = domain_get_by_id(domain, &entry);
      if (err_is_fail(err)) {
          err = b->tx_vtbl.get_status_response(b, NOP_CONT, pse, NULL, 0,
                                               err);
          if(err_is_fail(err)) {
              DEBUG_ERR(err, "status_response");
          }
      }
  
      pse.status = entry->status;
  
      err = b->tx_vtbl.get_status_response(b, NOP_CONT, pse, entry->argbuf, entry->argbytes,
                                           SYS_ERR_OK);
      if(err_is_fail(err)) {
          DEBUG_ERR(err, "status_response");
      }
  }
  
  static struct proc_mgmt_rx_vtbl monitor_vtbl = {
      .add_spawnd           = add_spawnd_handler,
      .spawn_call           = spawn_handler,
      .spawn_with_caps_call = spawn_with_caps_handler,
      .span_call            = span_handler,
      .kill_call            = kill_handler,
      .exit_call            = exit_handler,
      .wait_call            = wait_handler
  };
  
  static struct proc_mgmt_rx_vtbl non_monitor_vtbl = {
      .add_spawnd           = add_spawnd_handler_non_monitor,
      .spawn_call           = spawn_handler,
      .spawn_with_caps_call = spawn_with_caps_handler,
      .span_call            = span_handler,
      .kill_call            = kill_handler,
      .exit_call            = exit_handler,
      .wait_call            = wait_handler,
      .get_domainlist_call  = get_domainlist_handler,
      .get_status_call      = get_status_handler
  };
  
  /**
   * \brief Allocates a special LMP endpoint for authenticating with the monitor.
   */
  static errval_t alloc_ep_for_monitor(struct capref *ep)
  {
      struct proc_mgmt_lmp_binding *lmpb =
          malloc(sizeof(struct proc_mgmt_lmp_binding));
      assert(lmpb != NULL);
  
      // setup our end of the binding
      errval_t err = proc_mgmt_client_lmp_accept(lmpb, get_default_waitset(),
                                                 DEFAULT_LMP_BUF_WORDS);
      if (err_is_fail(err)) {
          free(lmpb);
          return err_push(err, LIB_ERR_PROC_MGMT_CLIENT_ACCEPT);
      }
  
      *ep = lmpb->chan.local_cap;
      lmpb->b.rx_vtbl = monitor_vtbl;
  
      return SYS_ERR_OK;
  }
  
  static void export_cb(void *st, errval_t err, iref_t iref)
  {
      if (err_is_fail(err)) {
          USER_PANIC_ERR(err, "export failed");
      }
  
      // Allocate an endpoint for the local monitor, who will use it to inform
      // us about new spawnd irefs on behalf of other monitors.
      struct capref ep;
      err = alloc_ep_for_monitor(&ep);
      if (err_is_fail(err)) {
          USER_PANIC_ERR(err, "failed to allocate LMP EP for local monitor");
      }
  
      // Send the endpoint to the monitor, so it can finish the handshake.
      struct monitor_binding *mb = get_monitor_binding();
      err = mb->tx_vtbl.set_proc_mgmt_ep_request(mb, NOP_CONT, ep);
      if (err_is_fail(err)) {
          USER_PANIC_ERR(err, "failed to send set_proc_mgmt_ep_request to "
                         "monitor");
      }
  
      // Also register this iref with the name service, for arbitrary client
      // domains to use for spawn-related ops.
      err = nameservice_register(SERVICE_BASENAME, iref);
      if (err_is_fail(err)) {
          USER_PANIC_ERR(err, "nameservice_register failed");
      }
  }
  
  static errval_t connect_cb(void *st, struct proc_mgmt_binding *b)
  {
      b->rx_vtbl = non_monitor_vtbl;
      return SYS_ERR_OK;
  }
  
  errval_t start_service(void)
  {
      errval_t err = domain_prealloc_caps();
      if (err_is_fail(err)) {
          USER_PANIC_ERR(err_push(err, PROC_MGMT_ERR_CREATE_DOMAIN_CAP),
                         "domain_prealloc_caps in start_service");
      }
  
      return proc_mgmt_export(NULL, export_cb, connect_cb, get_default_waitset(),
              IDC_EXPORT_FLAGS_DEFAULT);
  }
  