std::string stringify(const hashset<T>& set)
{
  std::ostringstream out;
  out << "{ ";
  typename hashset<T>::const_iterator iterator = set.begin();
  while (iterator != set.end()) {
    out << stringify(*iterator);
    if (++iterator != set.end()) {
      out << ", ";
    }
  }
  out << " }";
  return out.str();
}

static void
find_completions (
  drd_info drd, tree t, hashset<string>& h, string prefix= "")
{
  if (is_atomic (t)) {
    string s= t->label;
    int i= 0, n= N(s);
    while (i<n) {
      if (is_iso_alpha (s[i])) {
	int start= i;
	while ((i<n) && (is_iso_alpha (s[i]))) i++;
	string r= s (start, i);
	if (starts (r, prefix) && (r != prefix))
	  h->insert (r (N(prefix), N(r)));
      }
      else skip_symbol (s, i);
    }
  }
  else {
    int i, n= N(t);
    for (i=0; i<n; i++)
      if (drd->is_accessible_child (t, i))
	find_completions (drd, t[i], h, prefix);
  }
}

Try<Owned<VolumeManager>> VolumeManager::create(
    const http::URL& agentUrl,
    const string& rootDir,
    const CSIPluginInfo& info,
    const hashset<Service>& services,
    const string& containerPrefix,
    const Option<string>& authToken,
    Metrics* metrics)
{
  if (services.empty()) {
    return Error(
        "Must specify at least one service for CSI plugin type '" +
        info.type() + "' and name '" + info.name() + "'");
  }

  return new v0::VolumeManager(
      agentUrl,
      rootDir,
      info,
      services,
      containerPrefix,
      authToken,
      Runtime(),
      metrics);
}

  virtual Future<hashset<string>> watch(
      const hashset<string>& knownProfiles,
      const ResourceProviderInfo& resourceProviderInfo) override
  {
    // If the input set of profiles is empty, that means the caller is in sync
    // with this module. Hence, we return a future that will never be satisified
    // because this module will never return a non-empty set of profiles.
    if (knownProfiles.empty()) {
      return Future<hashset<string>>();
    }

    return hashset<string>::EMPTY;
  }

Future<Nothing> ExternalContainerizerProcess::__recover(
    const Option<state::SlaveState>& state,
    const hashset<ContainerID>& containers)
{
  VLOG(1) << "Recover continuation triggered";

  // An orphaned container is known to the external containerizer but
  // not to the slave, thus not recoverable but pending.
  hashset<ContainerID> orphaned = containers;

  if (state.isSome()) {
    foreachvalue (const FrameworkState& framework, state.get().frameworks) {
      foreachvalue (const ExecutorState& executor, framework.executors) {
        if (executor.info.isNone()) {
          LOG(WARNING) << "Skipping recovery of executor '" << executor.id
                       << "' of framework " << framework.id
                       << " because its info could not be recovered";
          continue;
        }

        if (executor.latest.isNone()) {
          LOG(WARNING) << "Skipping recovery of executor '" << executor.id
                       << "' of framework " << framework.id
                       << " because its latest run could not be recovered";
          continue;
        }

        // We are only interested in the latest run of the executor!
        const ContainerID& containerId = executor.latest.get();
        Option<RunState> run = executor.runs.get(containerId);
        CHECK_SOME(run);

        if (run.get().completed) {
          VLOG(1) << "Skipping recovery of executor '" << executor.id
                  << "' of framework " << framework.id
                  << " because its latest run "
                  << containerId << " is completed";
          continue;
        }

        // Containers the external containerizer does not have
        // information on, should be skipped as their state is not
        // recoverable.
        if (!containers.contains(containerId)) {
          LOG(WARNING) << "Skipping recovery of executor '" << executor.id
                       << "' of framework " << framework.id
                       << " because the external containerizer has not "
                       << " identified " << containerId << " as active";
          continue;
        }

        LOG(INFO) << "Recovering container '" << containerId
                  << "' for executor '" << executor.id
                  << "' of framework " << framework.id;

        // Re-create the sandbox for this container.
        const string& directory = paths::createExecutorDirectory(
            flags.work_dir,
            state.get().id,
            framework.id,
            executor.id,
            containerId);

        Option<string> user = None();
        if (flags.switch_user) {
          // The command (either in form of task or executor command)
          // can define a specific user to run as. If present, this
          // precedes the framework user value.
          if (executor.info.isSome() &&
              executor.info.get().command().has_user()) {
            user = executor.info.get().command().user();
          } else if (framework.info.isSome()) {
            user = framework.info.get().user();
          }
        }

        Sandbox sandbox(directory, user);

        // Collect this container as being active.
        actives.put(containerId, Owned<Container>(new Container(sandbox)));

        // Assume that this container had been launched, if this proves
        // to be wrong, the containerizer::Termination delivered by the
        // subsequent wait invocation will tell us.
        actives[containerId]->launched.set(Nothing());

        // Remove this container from the orphan collection as it is not
        // orphaned.
        orphaned.erase(containerId);
      }
    }
  }

void
cache_save (string buffer) {
  if (cache_changed->contains (buffer)) {
    url cache_file= texmacs_home_path * url ("system/cache/" * buffer);
    string cached;
    iterator<tree> it= iterate (cache_data);
    if (buffer == "file_cache" || buffer == "doc_cache") {
      while (it->busy ()) {
	tree ckey= it->next ();
	if (ckey[0] == buffer) {
	  cached << ckey[1]->label << "\n";
	  cached << cache_data [ckey]->label << "\n";
	  cached << "%-%-tm-cache-%-%\n";
	}
      }
    }
    else {
      cached << "(tuple\n";
      while (it->busy ()) {
	tree ckey= it->next ();
	if (ckey[0] == buffer) {
	  cached << tree_to_scheme (ckey[1]) << " ";
	  cached << tree_to_scheme (cache_data [ckey]) << "\n";
	}
      }
      cached << ")";
    }
    (void) save_string (cache_file, cached);
    cache_changed->remove (buffer);
  }
}

bool
xml_html_parser::build_valid_child (string parent, string child) {
  if (!html) return true;
  if ((parent == "<bottom>") || (parent == "html") || (parent == "body"))
    return true;
  if (html_empty_tag_table->contains (parent)) return false;
  if (!html_auto_close_table->contains (child)) return true;
  if (parent == "p") return !html_block_table->contains (child);
  if ((child == "dt") || (child == "dd")) return parent == "dl";
  if (child == "li")
    return (parent == "ul") || (parent == "ol") ||
           (parent == "dir") || (parent == "menu");
  if (child == "option") return (parent == "select") || (parent == "optgroup");
  if ((child == "thead") || (child == "tfoot") || (child == "tbody"))
    return parent == "table";
  if (child == "colgroup") return parent == "table";
  if (child == "col") return (parent == "table") || (parent == "colgroup");
  if (child == "tr")
    return (parent == "table") || (parent == "thead") ||
           (parent == "tfoot") || (parent == "tbody");
  if ((child == "th") || (child == "td"))
    return (parent == "tr") ||
           (parent == "table") || (parent == "thead") ||
           (parent == "tfoot") || (parent == "tbody");
  return true;
}

static void addtnormal(const vec &pos, int smooth, float offset, int normal1, int normal2, const vec &pos1, const vec &pos2)
{
    normalkey key = { pos, smooth };
    normalgroup &g = normalgroups.access(key, key);
    tnormal &n = tnormals.add();
    n.next = g.tnormals;
    n.offset = offset;
    n.normals[0] = normal1;
    n.normals[1] = normal2;
    normalkey key1 = { pos1, smooth }, key2 = { pos2, smooth };
    n.groups[0] = normalgroups.access(key1);
    n.groups[1] = normalgroups.access(key2);
    g.tnormals = tnormals.length()-1;
}

static int addnormal(const vec &pos, int smooth, int axis)
{
    normalkey key = { pos, smooth };
    normalgroup &g = normalgroups.access(key, key);
    g.flat += 1<<(4*axis);
    return axis - 6;
}

void
xml_html_parser::build (tree& r) {
  while (i<n) {
    if (is_tuple (a[i], "begin")) {
      string name= a[i][1]->label;
      if (build_must_close (name)) return;
      tree sub= copy (a[i]); sub[0]= "tag";
      i++;
      if (html && html_empty_tag_table->contains (name))
	r << sub;
      else {
	stack= tuple (name, stack);
	build (sub);
	r << sub;
	stack= stack[1];
      }
    }
    else if (is_tuple (a[i], "end")) {
      if (stack[0]->label == a[i][1]->label) { i++; return; }
      if (build_can_close (a[i][1]->label)) return;
      i++;
    }
    else r << a[i++];
  }
}

static int addnormal(const vec &key, const vec &surface)
{
    normalgroup &g = normalgroups.access(key, key);
    normal &n = normals.add();
    n.next = g.normals;
    n.surface = surface;
    return g.normals = normals.length()-1;
}

void
cache_set (string buffer, tree key, tree t) {
  tree ckey= tuple (buffer, key);
  if (cache_data[ckey] != t) {
    cache_data (ckey)= t;
    cache_changed->insert (buffer);
  }
}

pipe_link_rep::pipe_link_rep (string cmd2): cmd (cmd2) {
  pipe_link_set->insert ((pointer) this);
  in     = pp_in [0]= pp_in [1]= -1;
  out    = pp_out[0]= pp_out[1]= -1;
  err    = pp_err[0]= pp_err[1]= -1;
  outbuf = "";
  errbuf = "";
  alive  = false;
}

static void addtnormal(const vec &key, float offset, int normal1, int normal2, normalgroup *group1, normalgroup *group2)
{
    normalgroup &g = normalgroups.access(key, key);
    tnormal &n = tnormals.add();
    n.next = g.tnormals;
    n.offset = offset;
    n.normals[0] = normal1;
    n.normals[1] = normal2;
    n.groups[0] = group1;
    n.groups[1] = group2;
    g.tnormals = tnormals.length()-1;
}

socket_link_rep::socket_link_rep (string host2, int port2, int type2, int fd):
  host (host2), port (port2), type (type2)
{
  socket_link_set->insert ((pointer) this);
  io     = fd;
  outbuf = "";
  alive  = (fd != -1);
  if (type == SOCKET_SERVER) {
    sn = socket_notifier (io, &socket_callback, this, NULL);  
    add_notifier (sn);
    call ("server-add", object (io));
  }
}

static bool
is_thin (string s) {
  if (N(thin_delims) == 0)
    thin_delims << string ("|") << string ("||") << string ("interleave")
                << string ("[") << string ("]")
                << string ("lfloor") << string ("rfloor")
                << string ("lceil") << string ("rceil")
                << string ("llbracket") << string ("rrbracket")
                << string ("dlfloor") << string ("drfloor")
                << string ("dlceil") << string ("drceil")
                << string ("tlbracket") << string ("trbracket")
                << string ("tlfloor") << string ("trfloor")
                << string ("tlceil") << string ("trceil");
  return thin_delims->contains (s);
}

  foreach (const ContainerState& state, states) {
    expected.insert(state.container_id());

    if (!containers.contains(state.container_id())) {
      // The fact that we did not have a freezer (or systemd) cgroup
      // for this container implies this container has already been
      // destroyed but we need to add it to `containers` so that when
      // `LinuxLauncher::destroy` does get called below for this
      // container we will not fail.
      Container container;
      container.id = state.container_id();
      container.pid = state.pid();

      containers.put(container.id, container);

      LOG(INFO) << "Recovered (destroyed) container " << container.id;
    } else {
      // This container exists, so we save the pid so we can check
      // that it's part of the systemd "Mesos executor slice" below.
      containers[state.container_id()].pid = state.pid();
    }
  }

 void setteaminfo(const char *team, int frags)
 {
     teaminfo *t = teaminfos.access(team);
     if(!t) { t = &teaminfos[team]; copystring(t->team, team, sizeof(t->team)); }
     t->frags = frags;
 }

 void clearteaminfo()
 {
     teaminfos.clear();
 }

pipe_link_rep::~pipe_link_rep () {
  stop ();
  pipe_link_set->remove ((pointer) this);
}