text:
/* This is where the trie is built.  It's on-disk.  */
#include "lm/search_trie.hh"

#include "lm/blank.hh"
#include "lm/lm_exception.hh"
#include "lm/max_order.hh"
#include "lm/quantize.hh"
#include "lm/read_arpa.hh"
#include "lm/trie.hh"
#include "lm/vocab.hh"
#include "lm/weights.hh"
#include "lm/word_index.hh"
#include "util/ersatz_progress.hh"
#include "util/file_piece.hh"
#include "util/have.hh"
#include "util/proxy_iterator.hh"
#include "util/scoped.hh"

#include <algorithm>
#include <cmath>
#include <cstring>
#include <cstdio>
#include <deque>
#include <limits>
#include <numeric>
#include <vector>

#include <sys/mman.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <stdlib.h>
#include <unistd.h>

namespace lm {
namespace ngram {
namespace trie {
namespace {

/* An entry is a n-gram with probability.  It consists of:
 * WordIndex[order]
 * float probability
 * backoff probability (omitted for highest order n-gram)
 * These are stored consecutively in memory.  We want to sort them.
 *
 * The problem is the length depends on order (but all n-grams being compared
 * have the same order).  Allocating each entry on the heap (i.e. std::vector
 * or std::string) then sorting pointers is the normal solution.  But that's
 * too memory inefficient.  A lot of this code is just here to force std::sort
 * to work with records where length is specified at runtime (and avoid using
 * Boost for LM code).  I could have used qsort, but the point is to also
 * support __gnu_cxx:parallel_sort which doesn't have a qsort version.
 */

class EntryIterator {
  public:
    EntryIterator() {}

    EntryIterator(void *ptr, std::size_t size) : ptr_(static_cast<uint8_t*>(ptr)), size_(size) {}

    bool operator==(const EntryIterator &other) const {
      return ptr_ == other.ptr_;
    }
    bool operator<(const EntryIterator &other) const {
      return ptr_ < other.ptr_;
    }
    EntryIterator &operator+=(std::ptrdiff_t amount) {
      ptr_ += amount * size_;
      return *this;
    }
    std::ptrdiff_t operator-(const EntryIterator &other) const {
      return (ptr_ - other.ptr_) / size_;
    }

    const void *Data() const { return ptr_; }
    void *Data() { return ptr_; }
    std::size_t EntrySize() const { return size_; }

  private:
    uint8_t *ptr_;
    std::size_t size_;
};

class EntryProxy {
  public:
    EntryProxy() {}

    EntryProxy(void *ptr, std::size_t size) : inner_(ptr, size) {}

    operator std::string() const {
      return std::string(reinterpret_cast<const char*>(inner_.Data()), inner_.EntrySize());
    }

    EntryProxy &operator=(const EntryProxy &from) {
      memcpy(inner_.Data(), from.inner_.Data(), inner_.EntrySize());
      return *this;
    }

    EntryProxy &operator=(const std::string &from) {
      memcpy(inner_.Data(), from.data(), inner_.EntrySize());
      return *this;
    }

    const WordIndex *Indices() const {
      return reinterpret_cast<const WordIndex*>(inner_.Data());
    }

  private:
    friend class util::ProxyIterator<EntryProxy>;

    typedef std::string value_type;

    typedef EntryIterator InnerIterator;
    InnerIterator &Inner() { return inner_; }
    const InnerIterator &Inner() const { return inner_; }
    InnerIterator inner_;
};

typedef util::ProxyIterator<EntryProxy> NGramIter;

// Proxy for an entry except there is some extra cruft between the entries.  This is used to sort (n-1)-grams using the same memory as the sorted n-grams.
class PartialViewProxy {
  public:
    PartialViewProxy() : attention_size_(0), inner_() {}

    PartialViewProxy(void *ptr, std::size_t block_size, std::size_t attention_size) : attention_size_(attention_size), inner_(ptr, block_size) {}

    operator std::string() const {
      return std::string(reinterpret_cast<const char*>(inner_.Data()), attention_size_);
    }

    PartialViewProxy &operator=(const PartialViewProxy &from) {
      memcpy(inner_.Data(), from.inner_.Data(), attention_size_);
      return *this;
    }

    PartialViewProxy &operator=(const std::string &from) {
      memcpy(inner_.Data(), from.data(), attention_size_);
      return *this;
    }

    const WordIndex *Indices() const {
      return reinterpret_cast<const WordIndex*>(inner_.Data());
    }

  private:
    friend class util::ProxyIterator<PartialViewProxy>;

    typedef std::string value_type;

    const std::size_t attention_size_;

    typedef EntryIterator InnerIterator;
    InnerIterator &Inner() { return inner_; }
    const InnerIterator &Inner() const { return inner_; }
    InnerIterator inner_;
};

typedef util::ProxyIterator<PartialViewProxy> PartialIter;

template <class Proxy> class CompareRecords : public std::binary_function<const Proxy &, const Proxy &, bool> {
  public:
    explicit CompareRecords(unsigned char order) : order_(order) {}

    bool operator()(const Proxy &first, const Proxy &second) const {
      return Compare(first.Indices(), second.Indices());
    }
    bool operator()(const Proxy &first, const std::string &second) const {
      return Compare(first.Indices(), reinterpret_cast<const WordIndex*>(second.data()));
    }
    bool operator()(const std::string &first, const Proxy &second) const {
      return Compare(reinterpret_cast<const WordIndex*>(first.data()), second.Indices());
    }
    bool operator()(const std::string &first, const std::string &second) const {
      return Compare(reinterpret_cast<const WordIndex*>(first.data()), reinterpret_cast<const WordIndex*>(second.data()));
    }

  private:
    bool Compare(const WordIndex *first, const WordIndex *second) const {
      const WordIndex *end = first + order_;
      for (; first != end; ++first, ++second) {
        if (*first < *second) return true;
        if (*first > *second) return false;
      }
      return false;
    }

    unsigned char order_;
};

FILE *OpenOrThrow(const char *name, const char *mode) {
  FILE *ret = fopen(name, mode);
  if (!ret) UTIL_THROW(util::ErrnoException, "Could not open " << name << " for " << mode);
  return ret;
}

void WriteOrThrow(FILE *to, const void *data, size_t size) {
  assert(size);
  if (1 != std::fwrite(data, size, 1, to)) UTIL_THROW(util::ErrnoException, "Short write; requested size " << size);
}

void ReadOrThrow(FILE *from, void *data, size_t size) {
  if (1 != std::fread(data, size, 1, from)) UTIL_THROW(util::ErrnoException, "Short read; requested size " << size);
}

const std::size_t kCopyBufSize = 512;
void CopyOrThrow(FILE *from, FILE *to, size_t size) {
  char buf[std::min<size_t>(size, kCopyBufSize)];
  for (size_t i = 0; i < size; i += kCopyBufSize) {
    std::size_t amount = std::min(size - i, kCopyBufSize);
    ReadOrThrow(from, buf, amount);
    WriteOrThrow(to, buf, amount);
  }
}

void CopyRestOrThrow(FILE *from, FILE *to) {
  char buf[kCopyBufSize];
  size_t amount;
  while ((amount = fread(buf, 1, kCopyBufSize, from))) {
    WriteOrThrow(to, buf, amount);
  }
  if (!feof(from)) UTIL_THROW(util::ErrnoException, "Short read");
}

void RemoveOrThrow(const char *name) {
  if (std::remove(name)) UTIL_THROW(util::ErrnoException, "Could not remove " << name);
}

std::string DiskFlush(const void *mem_begin, const void *mem_end, const std::string &file_prefix, std::size_t batch, unsigned char order, std::size_t weights_size) {
  const std::size_t entry_size = sizeof(WordIndex) * order + weights_size;
  const std::size_t prefix_size = sizeof(WordIndex) * (order - 1);
  std::stringstream assembled;
  assembled << file_prefix << static_cast<unsigned int>(order) << '_' << batch;
  std::string ret(assembled.str());
  util::scoped_FILE out(OpenOrThrow(ret.c_str(), "w"));
  // Compress entries that being with the same (order-1) words.
  for (const uint8_t *group_begin = static_cast<const uint8_t*>(mem_begin); group_begin != static_cast<const uint8_t*>(mem_end);) {
    const uint8_t *group_end;
    for (group_end = group_begin + entry_size;
         (group_end != static_cast<const uint8_t*>(mem_end)) && !memcmp(group_begin, group_end, prefix_size);
         group_end += entry_size) {}
    WriteOrThrow(out.get(), group_begin, prefix_size);
    WordIndex group_size = (group_end - group_begin) / entry_size;
    WriteOrThrow(out.get(), &group_size, sizeof(group_size));
    for (const uint8_t *i = group_begin; i != group_end; i += entry_size) {
      WriteOrThrow(out.get(), i + prefix_size, sizeof(WordIndex));
      WriteOrThrow(out.get(), i + sizeof(WordIndex) * order, weights_size);
    }
    group_begin = group_end;
  }
  return ret;
}

class SortedFileReader {
  public:
    SortedFileReader() : ended_(false) {}

    void Init(const std::string &name, unsigned char order) {
      file_.reset(OpenOrThrow(name.c_str(), "r"));
      header_.resize(order - 1);
      NextHeader();
    }

    // Preceding words.
    const WordIndex *Header() const {
      return &*header_.begin();
    }
    const std::vector<WordIndex> &HeaderVector() const { return header_;}

    std::size_t HeaderBytes() const { return header_.size() * sizeof(WordIndex); }

    void NextHeader() {
      if (1 != fread(&*header_.begin(), HeaderBytes(), 1, file_.get())) {
        if (feof(file_.get())) {
          ended_ = true;
        } else {
          UTIL_THROW(util::ErrnoException, "Short read of counts");
        }
      }
    }

    WordIndex ReadCount() {
      WordIndex ret;
      ReadOrThrow(file_.get(), &ret, sizeof(WordIndex));
      return ret;
    }

    WordIndex ReadWord() {
      WordIndex ret;
      ReadOrThrow(file_.get(), &ret, sizeof(WordIndex));
      return ret;
    }

    template <class Weights> void ReadWeights(Weights &weights) {
      ReadOrThrow(file_.get(), &weights, sizeof(Weights));
    }

    bool Ended() const {
      return ended_;
    }

    void Rewind() {
      rewind(file_.get());
      ended_ = false;
      NextHeader();
    }

    FILE *File() { return file_.get(); }

  private:
    util::scoped_FILE file_;

    std::vector<WordIndex> header_;

    bool ended_;
};

void CopyFullRecord(SortedFileReader &from, FILE *to, std::size_t weights_size) {
  WriteOrThrow(to, from.Header(), from.HeaderBytes());
  WordIndex count = from.ReadCount();
  WriteOrThrow(to, &count, sizeof(WordIndex));

  CopyOrThrow(from.File(), to, (weights_size + sizeof(WordIndex)) * count);
}

void MergeSortedFiles(const std::string &first_name, const std::string &second_name, const std::string &out, std::size_t weights_size, unsigned char order) {
  SortedFileReader first, second;
  first.Init(first_name.c_str(), order);
  RemoveOrThrow(first_name.c_str());
  second.Init(second_name.c_str(), order);
  RemoveOrThrow(second_name.c_str());
  util::scoped_FILE out_file(OpenOrThrow(out.c_str(), "w"));
  while (!first.Ended() && !second.Ended()) {
    if (first.HeaderVector() < second.HeaderVector()) {
      CopyFullRecord(first, out_file.get(), weights_size);
      first.NextHeader();
      continue;
    }
    if (first.HeaderVector() > second.HeaderVector()) {
      CopyFullRecord(second, out_file.get(), weights_size);
      second.NextHeader();
      continue;
    }
    // Merge at the entry level.
    WriteOrThrow(out_file.get(), first.Header(), first.HeaderBytes());
    WordIndex first_count = first.ReadCount(), second_count = second.ReadCount();
    WordIndex total_count = first_count + second_count;
    WriteOrThrow(out_file.get(), &total_count, sizeof(WordIndex));

    WordIndex first_word = first.ReadWord(), second_word = second.ReadWord();
    WordIndex first_index = 0, second_index = 0;
    while (true) {
      if (first_word < second_word) {
        WriteOrThrow(out_file.get(), &first_word, sizeof(WordIndex));
        CopyOrThrow(first.File(), out_file.get(), weights_size);
        if (++first_index == first_count) break;
        first_word = first.ReadWord();
      } else {
        WriteOrThrow(out_file.get(), &second_word, sizeof(WordIndex));
        CopyOrThrow(second.File(), out_file.get(), weights_size);
        if (++second_index == second_count) break;
        second_word = second.ReadWord();
      }
    }
    if (first_index == first_count) {
      WriteOrThrow(out_file.get(), &second_word, sizeof(WordIndex));
      CopyOrThrow(second.File(), out_file.get(), (second_count - second_index) * (weights_size + sizeof(WordIndex)) - sizeof(WordIndex));
    } else {
      WriteOrThrow(out_file.get(), &first_word, sizeof(WordIndex));
      CopyOrThrow(first.File(), out_file.get(), (first_count - first_index) * (weights_size + sizeof(WordIndex)) - sizeof(WordIndex));
    }
    first.NextHeader();
    second.NextHeader();
  }

  for (SortedFileReader &remaining = first.Ended() ? second : first; !remaining.Ended(); remaining.NextHeader()) {
    CopyFullRecord(remaining, out_file.get(), weights_size);
  }
}

const char *kContextSuffix = "_contexts";

void WriteContextFile(uint8_t *begin, uint8_t *end, const std::string &ngram_file_name, std::size_t entry_size, unsigned char order) {
  const size_t context_size = sizeof(WordIndex) * (order - 1);
  // Sort just the contexts using the same memory.
  PartialIter context_begin(PartialViewProxy(begin + sizeof(WordIndex), entry_size, context_size));
  PartialIter context_end(PartialViewProxy(end + sizeof(WordIndex), entry_size, context_size));

  std::sort(context_begin, context_end, CompareRecords<PartialViewProxy>(order - 1));

  std::string name(ngram_file_name + kContextSuffix);
  util::scoped_FILE out(OpenOrThrow(name.c_str(), "w"));

  // Write out to file and uniqueify at the same time.  Could have used unique_copy if there was an appropriate OutputIterator.
  if (context_begin == context_end) return;
  PartialIter i(context_begin);
  WriteOrThrow(out.get(), i->Indices(), context_size);
  const WordIndex *previous = i->Indices();
  ++i;
  for (; i != context_end; ++i) {
    if (memcmp(previous, i->Indices(), context_size)) {
      WriteOrThrow(out.get(), i->Indices(), context_size);
      previous = i->Indices();
    }
  }
}

class ContextReader {
  public:
    ContextReader() : valid_(false) {}

    ContextReader(const char *name, unsigned char order) {
      Reset(name, order);
    }

    void Reset(const char *name, unsigned char order) {
      file_.reset(OpenOrThrow(name, "r"));
      length_ = sizeof(WordIndex) * static_cast<size_t>(order);
      words_.resize(order);
      valid_ = true;
      ++*this;
    }

    ContextReader &operator++() {
      if (1 != fread(&*words_.begin(), length_, 1, file_.get())) {
        if (!feof(file_.get()))
          UTIL_THROW(util::ErrnoException, "Short read");
        valid_ = false;
      }
      return *this;
    }

    const WordIndex *operator*() const { return &*words_.begin(); }

    operator bool() const { return valid_; }

    FILE *GetFile() { return file_.get(); }

  private:
    util::scoped_FILE file_;

    size_t length_;

    std::vector<WordIndex> words_;

    bool valid_;
};

void MergeContextFiles(const std::string &first_base, const std::string &second_base, const std::string &out_base, unsigned char order) {
  const size_t context_size = sizeof(WordIndex) * (order - 1);
  std::string first_name(first_base + kContextSuffix);
  std::string second_name(second_base + kContextSuffix);
  ContextReader first(first_name.c_str(), order - 1), second(second_name.c_str(), order - 1);
  RemoveOrThrow(first_name.c_str());
  RemoveOrThrow(second_name.c_str());
  std::string out_name(out_base + kContextSuffix);
  util::scoped_FILE out(OpenOrThrow(out_name.c_str(), "w"));
  while (first && second) {
    for (const WordIndex *f = *first, *s = *second; ; ++f, ++s) {
      if (f == *first + order - 1) {
        // Equal.
        WriteOrThrow(out.get(), *first, context_size);
        ++first;
        ++second;
        break;
      }
      if (*f < *s) {
        // First lower
        WriteOrThrow(out.get(), *first, context_size);
        ++first;
        break;
      } else if (*f > *s) {
        WriteOrThrow(out.get(), *second, context_size);
        ++second;
        break;
      }
    }
  }
  ContextReader &remaining = first ? first : second;
  if (!remaining) return;
  WriteOrThrow(out.get(), *remaining, context_size);
  CopyRestOrThrow(remaining.GetFile(), out.get());
}

void ConvertToSorted(util::FilePiece &f, const SortedVocabulary &vocab, const std::vector<uint64_t> &counts, util::scoped_memory &mem, const std::string &file_prefix, unsigned char order, PositiveProbWarn &warn) {
  ReadNGramHeader(f, order);
  const size_t count = counts[order - 1];
  // Size of weights.  Does it include backoff?
  const size_t words_size = sizeof(WordIndex) * order;
  const size_t weights_size = sizeof(float) + ((order == counts.size()) ? 0 : sizeof(float));
  const size_t entry_size = words_size + weights_size;
  const size_t batch_size = std::min(count, mem.size() / entry_size);
  uint8_t *const begin = reinterpret_cast<uint8_t*>(mem.get());
  std::deque<std::string> files;
  for (std::size_t batch = 0, done = 0; done < count; ++batch) {
    uint8_t *out = begin;
    uint8_t *out_end = out + std::min(count - done, batch_size) * entry_size;
    if (order == counts.size()) {
      for (; out != out_end; out += entry_size) {
        ReadNGram(f, order, vocab, reinterpret_cast<WordIndex*>(out), *reinterpret_cast<Prob*>(out + words_size), warn);
      }
    } else {
      for (; out != out_end; out += entry_size) {
        ReadNGram(f, order, vocab, reinterpret_cast<WordIndex*>(out), *reinterpret_cast<ProbBackoff*>(out + words_size), warn);
      }
    }
    // Sort full records by full n-gram.
    EntryProxy proxy_begin(begin, entry_size), proxy_end(out_end, entry_size);
    // parallel_sort uses too much RAM
    std::sort(NGramIter(proxy_begin), NGramIter(proxy_end), CompareRecords<EntryProxy>(order));
    files.push_back(DiskFlush(begin, out_end, file_prefix, batch, order, weights_size));
    WriteContextFile(begin, out_end, files.back(), entry_size, order);

    done += (out_end - begin) / entry_size;
  }

  // All individual files created.  Merge them.

  std::size_t merge_count = 0;
  while (files.size() > 1) {
    std::stringstream assembled;
    assembled << file_prefix << static_cast<unsigned int>(order) << "_merge_" << (merge_count++);
    files.push_back(assembled.str());
    MergeSortedFiles(files[0], files[1], files.back(), weights_size, order);
    MergeContextFiles(files[0], files[1], files.back(), order);
    files.pop_front();
    files.pop_front();
  }
  if (!files.empty()) {
    std::stringstream assembled;
    assembled << file_prefix << static_cast<unsigned int>(order) << "_merged";
    std::string merged_name(assembled.str());
    if (std::rename(files[0].c_str(), merged_name.c_str())) UTIL_THROW(util::ErrnoException, "Could not rename " << files[0].c_str() << " to " << merged_name.c_str());
    std::string context_name = files[0] + kContextSuffix;
    merged_name += kContextSuffix;
    if (std::rename(context_name.c_str(), merged_name.c_str())) UTIL_THROW(util::ErrnoException, "Could not rename " << context_name << " to " << merged_name.c_str());
  }
}

void ARPAToSortedFiles(const Config &config, util::FilePiece &f, std::vector<uint64_t> &counts, size_t buffer, const std::string &file_prefix, SortedVocabulary &vocab) {
  PositiveProbWarn warn(config.positive_log_probability);
  {
    std::string unigram_name = file_prefix + "unigrams";
    util::scoped_fd unigram_file;
    // In case <unk> appears.
    size_t extra_count = counts[0] + 1;
    util::scoped_mmap unigram_mmap(util::MapZeroedWrite(unigram_name.c_str(), extra_count * sizeof(ProbBackoff), unigram_file), extra_count * sizeof(ProbBackoff));
    Read1Grams(f, counts[0], vocab, reinterpret_cast<ProbBackoff*>(unigram_mmap.get()), warn);
    CheckSpecials(config, vocab);
    if (!vocab.SawUnk()) ++counts[0];
  }

  // Only use as much buffer as we need.
  size_t buffer_use = 0;
  for (unsigned int order = 2; order < counts.size(); ++order) {
    buffer_use = std::max<size_t>(buffer_use, static_cast<size_t>((sizeof(WordIndex) * order + 2 * sizeof(float)) * counts[order - 1]));
  }
  buffer_use = std::max<size_t>(buffer_use, static_cast<size_t>((sizeof(WordIndex) * counts.size() + sizeof(float)) * counts.back()));
  buffer = std::min<size_t>(buffer, buffer_use);

  util::scoped_memory mem;
  mem.reset(malloc(buffer), buffer, util::scoped_memory::MALLOC_ALLOCATED);
  if (!mem.get()) UTIL_THROW(util::ErrnoException, "malloc failed for sort buffer size " << buffer);

  for (unsigned char order = 2; order <= counts.size(); ++order) {
    ConvertToSorted(f, vocab, counts, mem, file_prefix, order, warn);
  }
  ReadEnd(f);
}

bool HeadMatch(const WordIndex *words, const WordIndex *const words_end, const WordIndex *header) {
  for (; words != words_end; ++words, ++header) {
    if (*words != *header) {
      //assert(*words <= *header);
      return false;
    }
  }
  return true;
}

// Phase to count n-grams, including blanks inserted because they were pruned but have extensions
class JustCount {
  public:
    template <class Middle, class Longest> JustCount(ContextReader * /*contexts*/, UnigramValue * /*unigrams*/, Middle * /*middle*/, Longest &/*longest*/, uint64_t *counts, unsigned char order)
      : counts_(counts), longest_counts_(counts + order - 1) {}

    void Unigrams(WordIndex begin, WordIndex end) {
      counts_[0] += end - begin;
    }

    void MiddleBlank(const unsigned char mid_idx, WordIndex /* idx */) {
      ++counts_[mid_idx + 1];
    }

    void Middle(const unsigned char mid_idx, const WordIndex * /*before*/, WordIndex /*key*/, const ProbBackoff &/*weights*/) {
      ++counts_[mid_idx + 1];
    }

    void Longest(WordIndex /*key*/, Prob /*prob*/) {
      ++*longest_counts_;
    }

    // Unigrams wrote one past.
    void Cleanup() {
      --counts_[0];
    }

  private:
    uint64_t *const counts_, *const longest_counts_;
};

// Phase to actually write n-grams to the trie.
template <class Quant> class WriteEntries {
  public:
    WriteEntries(ContextReader *contexts, UnigramValue *unigrams, BitPackedMiddle<typename Quant::Middle> *middle, BitPackedLongest<typename Quant::Longest> &longest, const uint64_t * /*counts*/, unsigned char order) :
      contexts_(contexts),
      unigrams_(unigrams),
      middle_(middle),
      longest_(longest),
      bigram_pack_((order == 2) ? static_cast<BitPacked&>(longest_) : static_cast<BitPacked&>(*middle_)) {}

    void Unigrams(WordIndex begin, WordIndex end) {
      uint64_t next = bigram_pack_.InsertIndex();
      for (UnigramValue *i = unigrams_ + begin; i < unigrams_ + end; ++i) {
        i->next = next;
      }
    }

    void MiddleBlank(const unsigned char mid_idx, WordIndex key) {
      middle_[mid_idx].Insert(key, kBlankProb, kBlankBackoff);
    }

    void Middle(const unsigned char mid_idx, const WordIndex *before, WordIndex key, ProbBackoff weights) {
      // Order (mid_idx+2).
      ContextReader &context = contexts_[mid_idx + 1];
      if (context && !memcmp(before, *context, sizeof(WordIndex) * (mid_idx + 1)) && (*context)[mid_idx + 1] == key) {
        SetExtension(weights.backoff);
        ++context;
      }
      middle_[mid_idx].Insert(key, weights.prob, weights.backoff);
    }

    void Longest(WordIndex key, Prob prob) {
      longest_.Insert(key, prob.prob);
    }

    void Cleanup() {}

  private:
    ContextReader *contexts_;
    UnigramValue *const unigrams_;
    BitPackedMiddle<typename Quant::Middle> *const middle_;
    BitPackedLongest<typename Quant::Longest> &longest_;
    BitPacked &bigram_pack_;
};

template <class Doing> class RecursiveInsert {
  public:
    template <class MiddleT, class LongestT> RecursiveInsert(SortedFileReader *inputs, ContextReader *contexts, UnigramValue *unigrams, MiddleT *middle, LongestT &longest, uint64_t *counts, unsigned char order) :
      doing_(contexts, unigrams, middle, longest, counts, order), inputs_(inputs), inputs_end_(inputs + order - 1), order_minus_2_(order - 2) {
    }

    // Outer unigram loop.
    void Apply(std::ostream *progress_out, const char *message, WordIndex unigram_count) {
      util::ErsatzProgress progress(progress_out, message, unigram_count + 1);
      for (words_[0] = 0; ; ++words_[0]) {
        progress.Set(words_[0]);
        WordIndex min_continue = unigram_count;
        for (SortedFileReader *other = inputs_; other != inputs_end_; ++other) {
          if (other->Ended()) continue;
          min_continue = std::min(min_continue, other->Header()[0]);
        }
        // This will write at unigram_count.  This is by design so that the next pointers will make sense.
        doing_.Unigrams(words_[0], min_continue + 1);
        if (min_continue == unigram_count) break;
        words_[0] = min_continue;
        Middle(0);
      }
      doing_.Cleanup();
    }

  private:
    void Middle(const unsigned char mid_idx) {
      // (mid_idx + 2)-gram.
      if (mid_idx == order_minus_2_) {
        Longest();
        return;
      }
      // Orders [2, order)

      SortedFileReader &reader = inputs_[mid_idx];

      if (reader.Ended() || !HeadMatch(words_, words_ + mid_idx + 1, reader.Header())) {
        // This order doesn't have a header match, but longer ones might.
        MiddleAllBlank(mid_idx);
        return;
      }

      // There is a header match.
      WordIndex count = reader.ReadCount();
      WordIndex current = reader.ReadWord();
      while (count) {
        WordIndex min_continue = std::numeric_limits<WordIndex>::max();
        for (SortedFileReader *other = inputs_ + mid_idx + 1; other < inputs_end_; ++other) {
          if (!other->Ended() && HeadMatch(words_, words_ + mid_idx + 1, other->Header()))
            min_continue = std::min(min_continue, other->Header()[mid_idx + 1]);
        }
        while (true) {
          if (current > min_continue) {
            doing_.MiddleBlank(mid_idx, min_continue);
            words_[mid_idx + 1] = min_continue;
            Middle(mid_idx + 1);
            break;
          }
          ProbBackoff weights;
          reader.ReadWeights(weights);
          doing_.Middle(mid_idx, words_, current, weights);
          --count;
          if (current == min_continue) {
            words_[mid_idx + 1] = min_continue;
            Middle(mid_idx + 1);
            if (count) current = reader.ReadWord();
            break;
          }
          if (!count) break;
          current = reader.ReadWord();
        }
      }
      // Count is now zero.  Finish off remaining blanks.
      MiddleAllBlank(mid_idx);
      reader.NextHeader();
    }

    void MiddleAllBlank(const unsigned char mid_idx) {
      while (true) {
        WordIndex min_continue = std::numeric_limits<WordIndex>::max();
        for (SortedFileReader *other = inputs_ + mid_idx + 1; other < inputs_end_; ++other) {
          if (!other->Ended() && HeadMatch(words_, words_ + mid_idx + 1, other->Header()))
            min_continue = std::min(min_continue, other->Header()[mid_idx + 1]);
        }
        if (min_continue == std::numeric_limits<WordIndex>::max()) return;
        doing_.MiddleBlank(mid_idx, min_continue);
        words_[mid_idx + 1] = min_continue;
        Middle(mid_idx + 1);
      }
    }

    void Longest() {
      SortedFileReader &reader = *(inputs_end_ - 1);
      if (reader.Ended() || !HeadMatch(words_, words_ + order_minus_2_ + 1, reader.Header())) return;
      WordIndex count = reader.ReadCount();
      for (WordIndex i = 0; i < count; ++i) {
        WordIndex word = reader.ReadWord();
        Prob prob;
        reader.ReadWeights(prob);
        doing_.Longest(word, prob);
      }
      reader.NextHeader();
      return;
    }

    Doing doing_;

    SortedFileReader *inputs_;
    SortedFileReader *inputs_end_;

    WordIndex words_[kMaxOrder];

    const unsigned char order_minus_2_;
};

void SanityCheckCounts(const std::vector<uint64_t> &initial, const std::vector<uint64_t> &fixed) {
  if (fixed[0] != initial[0]) UTIL_THROW(util::Exception, "Unigram count should be constant but initial is " << initial[0] << " and recounted is " << fixed[0]);
  if (fixed.back() != initial.back()) UTIL_THROW(util::Exception, "Longest count should be constant but it changed from " << initial.back() << " to " << fixed.back());
  for (unsigned char i = 0; i < initial.size(); ++i) {
    if (fixed[i] < initial[i]) UTIL_THROW(util::Exception, "Counts came out lower than expected.  This shouldn't happen");
  }
}

bool IsDirectory(const char *path) {
  struct stat info;
  if (0 != stat(path, &info)) return false;
  return S_ISDIR(info.st_mode);
}

template <class Quant> void TrainQuantizer(uint8_t order, uint64_t count, SortedFileReader &reader, util::ErsatzProgress &progress, Quant &quant) {
  ProbBackoff weights;
  std::vector<float> probs, backoffs;
  probs.reserve(count);
  backoffs.reserve(count);
  for (reader.Rewind(); !reader.Ended(); reader.NextHeader()) {
    uint64_t entries = reader.ReadCount();
    for (uint64_t c = 0; c < entries; ++c) {
      reader.ReadWord();
      reader.ReadWeights(weights);
      // kBlankProb isn't added yet.
      probs.push_back(weights.prob);
      if (weights.backoff != 0.0) backoffs.push_back(weights.backoff);
      ++progress;
    }
  }
  quant.Train(order, probs, backoffs);
}

template <class Quant> void TrainProbQuantizer(uint8_t order, uint64_t count, SortedFileReader &reader, util::ErsatzProgress &progress, Quant &quant) {
  Prob weights;
  std::vector<float> probs, backoffs;
  probs.reserve(count);
  for (reader.Rewind(); !reader.Ended(); reader.NextHeader()) {
    uint64_t entries = reader.ReadCount();
    for (uint64_t c = 0; c < entries; ++c) {
      reader.ReadWord();
      reader.ReadWeights(weights);
      // kBlankProb isn't added yet.
      probs.push_back(weights.prob);
      ++progress;
    }
  }
  quant.TrainProb(order, probs);
}

} // namespace

template <class Quant> void BuildTrie(const std::string &file_prefix, std::vector<uint64_t> &counts, const Config &config, TrieSearch<Quant> &out, Quant &quant, Backing &backing) {
  std::vector<SortedFileReader> inputs(counts.size() - 1);
  std::vector<ContextReader> contexts(counts.size() - 1);

  for (unsigned char i = 2; i <= counts.size(); ++i) {
    std::stringstream assembled;
    assembled << file_prefix << static_cast<unsigned int>(i) << "_merged";
    inputs[i-2].Init(assembled.str(), i);
    RemoveOrThrow(assembled.str().c_str());
    assembled << kContextSuffix;
    contexts[i-2].Reset(assembled.str().c_str(), i-1);
    RemoveOrThrow(assembled.str().c_str());
  }

  std::vector<uint64_t> fixed_counts(counts.size());
  {
    RecursiveInsert<JustCount> counter(&*inputs.begin(), &*contexts.begin(), NULL, out.middle_begin_, out.longest, &*fixed_counts.begin(), counts.size());
    counter.Apply(config.messages, "Counting n-grams that should not have been pruned", counts[0]);
  }
  for (std::vector<SortedFileReader>::const_iterator i = inputs.begin(); i != inputs.end(); ++i) {
    if (!i->Ended()) UTIL_THROW(FormatLoadException, "There's a bug in the trie implementation: the " << (i - inputs.begin() + 2) << "-gram table did not complete reading");
  }
  SanityCheckCounts(counts, fixed_counts);
  counts = fixed_counts;

  out.SetupMemory(GrowForSearch(config, TrieSearch<Quant>::Size(fixed_counts, config), backing), fixed_counts, config);

  if (Quant::kTrain) {
    util::ErsatzProgress progress(config.messages, "Quantizing", std::accumulate(counts.begin() + 1, counts.end(), 0));
    for (unsigned char i = 2; i < counts.size(); ++i) {
      TrainQuantizer(i, counts[i-1], inputs[i-2], progress, quant);
    }
    TrainProbQuantizer(counts.size(), counts.back(), inputs[counts.size() - 2], progress, quant);
    quant.FinishedLoading(config);
  }

  for (unsigned char i = 2; i <= counts.size(); ++i) {
    inputs[i-2].Rewind();
  }
  UnigramValue *unigrams = out.unigram.Raw();
  // Fill entries except unigram probabilities.
  {
    RecursiveInsert<WriteEntries<Quant> > inserter(&*inputs.begin(), &*contexts.begin(), unigrams, out.middle_begin_, out.longest, &*fixed_counts.begin(), counts.size());
    inserter.Apply(config.messages, "Building trie", fixed_counts[0]);
  }

  // Fill unigram probabilities.
  try {
    std::string name(file_prefix + "unigrams");
    util::scoped_FILE file(OpenOrThrow(name.c_str(), "r"));
    for (WordIndex i = 0; i < counts[0]; ++i) {
      ReadOrThrow(file.get(), &unigrams[i].weights, sizeof(ProbBackoff));
      if (contexts[0] && **contexts[0] == i) {
        SetExtension(unigrams[i].weights.backoff);
        ++contexts[0];
      }
    }
    RemoveOrThrow(name.c_str());
  } catch (util::Exception &e) {
    e << " while re-reading unigram probabilities";
    throw;
  }

  // Do not disable this error message or else too little state will be returned.  Both WriteEntries::Middle and returning state based on found n-grams will need to be fixed to handle this situation.
  for (unsigned char order = 2; order <= counts.size(); ++order) {
    const ContextReader &context = contexts[order - 2];
    if (context) {
      FormatLoadException e;
      e << "An " << static_cast<unsigned int>(order) << "-gram has the context (i.e. all but the last word):";
      for (const WordIndex *i = *context; i != *context + order - 1; ++i) {
        e << ' ' << *i;
      }
      e << " so this context must appear in the model as a " << static_cast<unsigned int>(order - 1) << "-gram but it does not";
      throw e;
    }
  }

  /* Set ending offsets so the last entry will be sized properly */
  // Last entry for unigrams was already set.
  if (out.middle_begin_ != out.middle_end_) {
    for (typename TrieSearch<Quant>::Middle *i = out.middle_begin_; i != out.middle_end_ - 1; ++i) {
      i->FinishedLoading((i+1)->InsertIndex());
    }
    (out.middle_end_ - 1)->FinishedLoading(out.longest.InsertIndex());
  }
}

template <class Quant> uint8_t *TrieSearch<Quant>::SetupMemory(uint8_t *start, const std::vector<uint64_t> &counts, const Config &config) {
  quant_.SetupMemory(start, config);
  start += Quant::Size(counts.size(), config);
  unigram.Init(start);
  start += Unigram::Size(counts[0]);
  FreeMiddles();
  middle_begin_ = static_cast<Middle*>(malloc(sizeof(Middle) * (counts.size() - 2)));
  middle_end_ = middle_begin_ + (counts.size() - 2);
  std::vector<uint8_t*> middle_starts(counts.size() - 2);
  for (unsigned char i = 2; i < counts.size(); ++i) {
    middle_starts[i-2] = start;
    start += Middle::Size(Quant::MiddleBits(config), counts[i-1], counts[0], counts[i]);
  }
  // Crazy backwards thing so we initialize in the correct order.
  for (unsigned char i = counts.size() - 1; i >= 2; --i) {
    new (middle_begin_ + i - 2) Middle(
        middle_starts[i-2],
        quant_.Mid(i),
        counts[0],
        counts[i],
        (i == counts.size() - 1) ? static_cast<const BitPacked&>(longest) : static_cast<const BitPacked &>(middle_begin_[i-1]));
  }
  longest.Init(start, quant_.Long(counts.size()), counts[0]);
  return start + Longest::Size(Quant::LongestBits(config), counts.back(), counts[0]);
}

template <class Quant> void TrieSearch<Quant>::LoadedBinary() {
  unigram.LoadedBinary();
  for (Middle *i = middle_begin_; i != middle_end_; ++i) {
    i->LoadedBinary();
  }
  longest.LoadedBinary();
}

template <class Quant> void TrieSearch<Quant>::InitializeFromARPA(const char *file, util::FilePiece &f, std::vector<uint64_t> &counts, const Config &config, SortedVocabulary &vocab, Backing &backing) {
  std::string temporary_directory;
  if (config.temporary_directory_prefix) {
    temporary_directory = config.temporary_directory_prefix;
    if (!temporary_directory.empty() && temporary_directory[temporary_directory.size() - 1] != '/' && IsDirectory(temporary_directory.c_str()))
      temporary_directory += '/';
  } else if (config.write_mmap) {
    temporary_directory = config.write_mmap;
  } else {
    temporary_directory = file;
  }
  // Null on end is kludge to ensure null termination.
  temporary_directory += "_trie_tmp_XXXXXX";
  temporary_directory += '\0';
  if (!mkdtemp(&temporary_directory[0])) {
    UTIL_THROW(util::ErrnoException, "Failed to make a temporary directory based on the name " << temporary_directory.c_str());
  }
  // Chop off null kludge.
  temporary_directory.resize(strlen(temporary_directory.c_str()));
  // Add directory delimiter.  Assumes a real operating system.
  temporary_directory += '/';
  // At least 1MB sorting memory.
  ARPAToSortedFiles(config, f, counts, std::max<size_t>(config.building_memory, 1048576), temporary_directory.c_str(), vocab);

  BuildTrie(temporary_directory, counts, config, *this, quant_, backing);
  if (rmdir(temporary_directory.c_str()) && config.messages) {
    *config.messages << "Failed to delete " << temporary_directory << std::endl;
  }
}

template class TrieSearch<DontQuantize>;
template class TrieSearch<SeparatelyQuantize>;

} // namespace trie
} // namespace ngram
} // namespace lm


text:
require 'tk'
require 'tk/frame'
require 'tkextlib/bwidget.rb'

module Tk
  module BWidget
    class StatusBar < TkWindow
    end
  end
end

class Tk::BWidget::StatusBar
  TkCommandNames = ['StatusBar'.freeze].freeze
  WidgetClassName = 'StatusBar'.freeze
  WidgetClassNames[WidgetClassName] = self

  def __boolval_optkeys
    super() << 'showresize'
  end
  private :__boolval_optkeys

  def add(win, keys={})
    tk_send('add', win, keys)
    self
  end

  def delete(*wins)
    tk_send('delete', *wins)
    self
  end

  def get_frame(&b)
    win = window(tk_send_without_enc('getframe'))
    win.instance_eval(&b) if b
    win
  end

  def items
    list(tk_send('items'))
  end
end

redpajama_set_name: RedPajamaGithub


text:
"""
   Copyright 2014 The Trustees of Princeton University

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
"""

# utility functions for making (gateway) watchdogs

import sys
import os
import grp
import pwd
import signal
import errno
import time
import logging
import psutil

logging.basicConfig( format='[%(levelname)s] [%(module)s:%(lineno)d] %(message)s' )

log = logging.getLogger()

#-------------------------------
def is_attr_str( attr_str ):
   return attr_str.startswith("attr:") and "=" in attr_str

#-------------------------------
def attr_proc_title( binary, attrs ):
   """
   Make a process title that has attr:<k>=<v> for set of attributes.
   """
   return "%s %s" % (binary, " ".join( ["attr:%s=%s" % (k, v) for (k, v) in attrs.items()] ))

#-------------------------------
def parse_proc_attrs( proc_cmdline ):
   """
   Given a process command-line, find out its attrs.
   """
   attrs = {}

   # format: <binary> <attr:k=v>...
   parts = proc_cmdline.split(" ")

   attr_list = parts[1:]

   for attr_kv in attr_list:

      if is_attr_str( attr_kv ):

         kv = attr_kv[len("atttr:")]
         kv_parts = kv.split("=")

         key = kv_parts[0]
         value = "=".join(kv_parts[1:])

         attrs[key] = value

   return attrs

#-------------------------------
def get_proc_pid( p ):
   return p.pid

#-------------------------------
def get_proc_cmdline( p ):
   # depending on which version of psutil we're using, this is either a list
   # or a callable that evaluates to a list.
   cmdline = None

   try:
      cmdline = p.cmdline()
   except:
      cmdline = p.cmdline

   assert type(cmdline) == list, "Unable to iterate over process command-lines"

   return cmdline

#-------------------------------
def find_by_attrs( watchdog_name, attrs ):
   """
   Find a running watchdog by a given attribute set
   NOTE: it must have a process cmdline generated by attr_proc_title()
   """
   procs = []
   for p in psutil.process_iter():

      cmdline = get_proc_cmdline( p )

      if len(cmdline) == 0:
         continue

      parts = cmdline[0].split(" ")
      if len(parts) < 2:
         continue

      name = parts[0]
      attr_list = parts[1:]

      if name == watchdog_name:

         # switch on attrs
         matching_attrs = []
         match = True

         # find matching attrs
         for attr_kv in attr_list:

            if is_attr_str( attr_kv ):
               # parse this: attr:<name>=<value>
               kv = attr_kv[len("attr:"):]

               kv_parts = kv.split("=")
               key = kv_parts[0]
               value = "=".join(kv_parts[1:])

               if attrs.has_key(key) and attrs[key] == value:
                  matching_attrs.append( key )

         # did we match all attrs?
         for attr_name in attrs.keys():
            if attr_name not in matching_attrs:
               match = False
               break

         if match:
            procs.append( p )

   return procs

#-------------------------------
def release_privileges( uid_name, gid_name ):

   if os.getuid() != 0:
      # not root, so nothing to do
      return 0

   # get desired uid/gid
   try:
      running_pwd = pwd.getpwnam( uid_name )
      running_uid = running_pwd.pw_uid
   except KeyError:
      log.error("No such user '%s'" % uid_name)
      return -errno.EINVAL

   try:
      running_grp = grp.getgrnam( gid_name )
      running_gid = running_grp.gr_gid
   except KeyError:
      log.error("No such group '%s'" % gid_name)
      return -errno.EINVAL

   # unset groups
   os.setgroups( [] )

   # set new uid/gid
   os.setgid( running_gid )
   os.setuid( running_uid )

   return 0

#-------------------------------
def spawn( child_method, stdin_method, old_exit_status, parent_signals=None, uid_name=None, gid_name=None ):
   # get stdin
   stdin_buf = None
   child_stdin = -1
   parent_w = -1

   if stdin_method is not None:
      stdin_buf = stdin_method()

      # create child stdin
      try:
         child_stdin, parent_w = os.pipe()
      except OSError, oe:
         print >> sys.stderr, "pipe errno = %s" % oe.errno
         return -oe.errno

   # fork, become a watchdog, and run the main server
   try:
      child_pid = os.fork()
   except OSError, oe:
      # failure
      log.exception(oe)
      log.error("Failed to fork, errno = %s" % oe.errno)

   if child_pid > 0:
      # we're the parent

      if parent_w != -1 and child_stdin != -1:
         # send stdin
         os.close( child_stdin )

         # parent.  Send stdin
         os.write( parent_w, stdin_buf )
         os.close( parent_w )

      log.info("Spawned child %s" % child_pid)

      return child_pid

   else:
      # we're the child...
      # set default signal handlers--don't mess with the parent's
      if parent_signals is not None:
         for sig in parent_signals:
            signal.signal( sig, signal.SIG_DFL )

      if parent_w != -1 and child_stdin != -1:
         os.close( parent_w )

         sys.stdout.flush()
         sys.stderr.flush()

         # redirect stdin
         os.dup2( child_stdin, sys.stdin.fileno() )
         os.close( child_stdin )

      # drop privileges, if we have to
      if uid_name is not None and gid_name is not None:
         release_privileges( uid_name, gid_name )

      rc = child_method( old_exit_status )
      sys.exit(rc)

#-------------------------------
def check_exit_status( exit_status, respawn_exit_statuses=None, respawn_signals=None ):
   # determine whether or not we need to respawn, based on the exit and signal status.

   need_respawn = False

   # did the child exit on its own accord?
   if os.WIFEXITED(exit_status):
      ex_status = os.WEXITSTATUS( exit_status )
      log.info("Child exit status %s" % ex_status )

      if respawn_exit_statuses is None or ex_status in respawn_exit_statuses:
         need_respawn = True

   # was the child signaled?
   elif os.WIFSIGNALED(exit_status):
      sig = os.WTERMSIG( exit_status )
      log.info("Child exited with signal %s" % sig )

      if respawn_signals is None or sig in respawn_signals:
         need_respawn = True

   return need_respawn

#-------------------------------
def kill_child( child_pid ):
   # send SIGKILL
   log.info("Sending SIGKILL to %s" % child_pid)
   os.kill( child_pid, signal.SIGKILL )

#-------------------------------
def stop_child( child_pid ):
   # send SIGTERM, then SIGKILL
   log.info("Sending SIGTERM to %s" % child_pid)
   os.kill( child_pid, signal.SIGTERM )

   count = 3
   while count > 0:
      try:
         pid, exit_status = os.waitpid( child_pid, os.WNOHANG )
      except OSError, e:
         log.exception(e)

         # if this wasn't ECHILD (i.e. the PID isn't invalid), then kill the process and die
         if e.errno != errno.ECHILD:
            kill_child( child_pid )
            return -1

      # did it exit?
      if os.WIFEXITED(exit_status) or os.WIFSIGNALED(exit_status):
         log.info("Child %s exited with status %s" % (child_pid, exit_status))
         break

      # if not, try again
      time.sleep(1.0)
      count -= 1

   if count == 0:
      kill_child( child_pid )

   return 0

#-------------------------------
def flap_wait( last_spawn, flap_delay, flap_threshold, flap_reset ):
   # wait before spawning another process, to avoid flapping.
   # return the new flap delay.
   t = time.time()
   if t - last_spawn > flap_reset:
      flap_delay = 1

   if t - last_spawn < flap_threshold:
      log.warning("Child is respawning too quickly.  Waiting %s seconds to try again" % flap_delay)

      time.sleep( flap_delay )

      flap_delay *= 2

      if flap_delay > flap_reset:
         flap_delay = flap_reset

   return flap_delay

#-------------------------------
def run( binary, argv, stdin_buf, uid_name=None, gid_name=None ):
   """
   Fork a child and feed it a stdin.
   Return its pid on success
   return errno on error
   """

   def do_exec( ignored ):
      # exec the binary
      try:
         os.execvp( binary, argv )
      except OSError, oe:
         log.error("Failed to execute %s, errno = %s" % (binary, -oe.errno))
         return -oe.errno

   stdin_method = lambda: stdin_buf

   return spawn( do_exec, stdin_method, 0, uid_name=uid_name, gid_name=gid_name )

#-------------------------------
def main( spawn_method, pid_cb=None, stdin_cb=None, exit_cb=None, respawn_exit_statuses=None, respawn_signals=None, parent_signals=None, flap_threshold=600, flap_reset=3600, uid_name=None, gid_name=None ):

   # fork, become a watchdog, and run the main server
   last_spawn = time.time()
   flap_delay = 1

   # but first, drop privileges if we need to
   if uid_name is not None and gid_name is not None:
      release_privileges( uid_name, gid_name )

   child_pid = spawn( spawn_method, stdin_cb, 0, parent_signals=parent_signals, uid_name=uid_name, gid_name=gid_name )

   if child_pid < 0:
      # spawn failed
      log.error("Failed to spawn, rc = %s" % child_pid)
      return child_pid

   if pid_cb is not None:
      # pass along the PID
      pid_cb( child_pid )

   # we're the parent...start monitoring
   while True:

      # wait for child to die
      try:
         pid, exit_status = os.waitpid( child_pid, 0 )
      except OSError, e:
         log.exception(e)
         log.error( "PID %s exiting" % os.getpid() )
         sys.exit(1)

      except (KeyboardInterrupt, SystemExit):
         # kill the child and return
         my_exit_status = stop_child( child_pid )
         log.info("watchdog exit, child exit status %s" % my_exit_status )
         return my_exit_status

      need_respawn = False

      # find out if we need to respawn
      if exit_cb is not None:
         need_respawn = exit_cb( exit_status )

      else:
         need_respawn = check_exit_status( exit_status, respawn_exit_statuses, respawn_signals )

      if need_respawn:

         # are we flapping?  Wait for a bit if so
         if flap_threshold > 0:
            flap_delay = flap_wait( last_spawn, flap_delay, flap_threshold, flap_reset )

         child_pid = spawn( spawn_method, stdin_cb, exit_status, parent_signals=parent_signals )

         if child_pid < 0:
            # spawn failed
            log.error("Failed to spawn, rc = %s" % child_pid)
            return child_pid

         # pass along the PID
         if pid_cb is not None:
            pid_cb( child_pid )

         last_spawn = time.time()

      else:
         # no respawn...return the exit status
         log.info("not respawning, child exit status %s" % exit_status )
         return exit_status


text:
<?php
namespace Feeds\Forms;

use Zend\Form\Element;
use Zend\Form\Form;

class SubscribeForm extends Form
{
    public function __construct($name = null)
    {
        parent::__construct('feeds-subscribe');

        $this->setAttribute('method', 'post');
        $this->setAttribute('class', 'input-append');

        $this->add(array(
            'name' => 'url',
            'type'  => 'Zend\Form\Element\Url',
            'attributes' => array(
                'class' => 'input-medium',
                'placeholder' => 'http://feeds.feedbu...'
            ),
        ));
        $this->add(new Element\Csrf('csrf'));
        $this->add(array(
            'name' => 'subscribe',
            'attributes' => array(
                'type'  => 'submit',
                'class' => 'btn btn-primary'
            ),
        ));
    }
}


text:
extern "C" IMAGE_DOS_HEADER __ImageBase;

namespace trace {
namespace client {

namespace {

class GetInstanceIdForModuleTest : public testing::Test {
 public:
  GetInstanceIdForModuleTest() : path_(L"C:\\path\\foo.exe") { }

  virtual void SetUp() override {
    testing::Test::SetUp();
    env_.reset(base::Environment::Create());
  }

  void SetEnvVar(const base::StringPiece& string) {
    ASSERT_TRUE(env_->SetVar(::kSyzygyRpcInstanceIdEnvVar, string.as_string()));
  }

  void UnsetEnvVar() {
    ASSERT_TRUE(env_->UnSetVar(::kSyzygyRpcInstanceIdEnvVar));
  }

  base::FilePath path_;
  scoped_ptr<base::Environment> env_;
};

class IsRpcSessionMandatoryTest : public testing::Test {
 public:
  IsRpcSessionMandatoryTest() : path_(L"C:\\path\\foo.exe") { }

  virtual void SetUp() override {
    testing::Test::SetUp();
    env_.reset(base::Environment::Create());
  }

  void SetEnvVar(const base::StringPiece& string) {
    ASSERT_TRUE(env_->SetVar(::kSyzygyRpcSessionMandatoryEnvVar,
                             string.as_string()));
  }

  void UnsetEnvVar() {
    ASSERT_TRUE(env_->UnSetVar(::kSyzygyRpcSessionMandatoryEnvVar));
  }

  base::FilePath path_;
  scoped_ptr<base::Environment> env_;
};

}  // namespace

TEST(GetModuleBaseAddressTest, WorksOnSelf) {
  void* module_base = NULL;
  EXPECT_TRUE(GetModuleBaseAddress(&GetModuleBaseAddress, &module_base));
  EXPECT_EQ(&__ImageBase, module_base);
}

TEST(GetModulePath, WorksOnSelf) {
  void* module_base = NULL;
  ASSERT_TRUE(GetModuleBaseAddress(&GetModuleBaseAddress, &module_base));

  base::FilePath module_path;
  EXPECT_TRUE(GetModulePath(module_base, &module_path));

  base::FilePath self_path =
      ::testing::GetExeRelativePath(L"rpc_client_lib_unittests.exe");
  EXPECT_SAME_FILE(self_path, module_path);
}

TEST_F(GetInstanceIdForModuleTest, ReturnsEmptyForNoEnvVar) {
  ASSERT_NO_FATAL_FAILURE(UnsetEnvVar());
  EXPECT_EQ(std::string(), GetInstanceIdForModule(path_));
}

TEST_F(GetInstanceIdForModuleTest, ReturnsEmptyForEmptyEnvVar) {
  ASSERT_NO_FATAL_FAILURE(SetEnvVar(""));
  EXPECT_EQ(std::string(), GetInstanceIdForModule(path_));
}

TEST_F(GetInstanceIdForModuleTest, ReturnsEmptyForNoMatch) {
  ASSERT_NO_FATAL_FAILURE(SetEnvVar("bar.exe,1;baz.exe,2"));
  EXPECT_EQ(std::string(""), GetInstanceIdForModule(path_));
}

TEST_F(GetInstanceIdForModuleTest, ReturnsGenericIdWhenNoPathMatches) {
  ASSERT_NO_FATAL_FAILURE(SetEnvVar("1 ;bar.exe,2"));
  EXPECT_EQ(std::string("1"), GetInstanceIdForModule(path_));
}

TEST_F(GetInstanceIdForModuleTest, ReturnsBaseNameId) {
  ASSERT_NO_FATAL_FAILURE(SetEnvVar("1; foo.exe , 2"));
  EXPECT_EQ(std::string("2"), GetInstanceIdForModule(path_));
}

TEST_F(GetInstanceIdForModuleTest, ReturnsExactPathId) {
  ASSERT_NO_FATAL_FAILURE(SetEnvVar("1;foo.exe,2;C:\\path\\foo.exe, 3 "));
  EXPECT_EQ(std::string("3"), GetInstanceIdForModule(path_));
}

TEST(GetInstanceIdForThisModuleTest, WorksAsExpected) {
  base::FilePath self_path =
      ::testing::GetExeRelativePath(L"rpc_client_lib_unittests.exe");

  std::wstring env_var(self_path.value());
  env_var.append(L",1");

  scoped_ptr<base::Environment> env;
  env.reset(base::Environment::Create());
  ASSERT_TRUE(env->SetVar(::kSyzygyRpcInstanceIdEnvVar,
                          base::WideToUTF8(env_var)));

  EXPECT_EQ(std::string("1"), GetInstanceIdForThisModule());
}

TEST_F(IsRpcSessionMandatoryTest, ReturnsFalseForNoEnvVar) {
  ASSERT_NO_FATAL_FAILURE(UnsetEnvVar());
  EXPECT_FALSE(IsRpcSessionMandatory(path_));
}

TEST_F(IsRpcSessionMandatoryTest, ReturnsFalseForEmptyEnvVar) {
  ASSERT_NO_FATAL_FAILURE(SetEnvVar(""));
  EXPECT_FALSE(IsRpcSessionMandatory(path_));
}

TEST_F(IsRpcSessionMandatoryTest, ReturnsFalseForNoMatch) {
  ASSERT_NO_FATAL_FAILURE(SetEnvVar("bar.exe,1;baz.exe,1"));
  EXPECT_FALSE(IsRpcSessionMandatory(path_));
}

TEST_F(IsRpcSessionMandatoryTest, ReturnsGlobalValueWhenNoPathMatches) {
  ASSERT_NO_FATAL_FAILURE(SetEnvVar("1 ; bar.exe,0"));
  EXPECT_TRUE(IsRpcSessionMandatory(path_));
}

TEST_F(IsRpcSessionMandatoryTest, ReturnsBaseNameValue) {
  ASSERT_NO_FATAL_FAILURE(SetEnvVar("0; foo.exe , 1"));
  EXPECT_TRUE(IsRpcSessionMandatory(path_));
}

TEST_F(IsRpcSessionMandatoryTest, ReturnsExactPathValue) {
  ASSERT_NO_FATAL_FAILURE(SetEnvVar("0;foo.exe,0;C:\\path\\foo.exe, 1 "));
  EXPECT_TRUE(IsRpcSessionMandatory(path_));
}

TEST_F(IsRpcSessionMandatoryTest, NonNumericIgnored) {
  ASSERT_NO_FATAL_FAILURE(SetEnvVar("foo.exe,baz;C:\\path\\foo.exe,bar"));
  EXPECT_FALSE(IsRpcSessionMandatory(path_));
}

TEST(IsRpcSessionMandatoryThisModuleTest, WorksAsExpected) {
  base::FilePath self_path =
      ::testing::GetExeRelativePath(L"rpc_client_lib_unittests.exe");

  std::wstring env_var(self_path.value());
  env_var.append(L",1");

  scoped_ptr<base::Environment> env;
  env.reset(base::Environment::Create());
  ASSERT_TRUE(env->SetVar(::kSyzygyRpcSessionMandatoryEnvVar,
                          base::WideToUTF8(env_var)));

  EXPECT_TRUE(IsRpcSessionMandatoryForThisModule());
}

TEST(InitializeRpcSessionTest, FailureSessionNotMandatory) {
  base::FilePath self_path =
      ::testing::GetExeRelativePath(L"rpc_client_lib_unittests.exe");

  scoped_ptr<base::Environment> env;
  env.reset(base::Environment::Create());

  std::wstring env_var(self_path.value());
  env_var.append(L",0");
  ASSERT_TRUE(env->SetVar(::kSyzygyRpcSessionMandatoryEnvVar,
                          base::WideToUTF8(env_var)));

  env_var = self_path.value();
  std::wstring id(L"dummy-id");
  env_var.append(L",");
  env_var.append(id);
  ASSERT_TRUE(env->SetVar(::kSyzygyRpcInstanceIdEnvVar,
                          base::WideToUTF8(env_var)));

  RpcSession session;
  TraceFileSegment segment;
  EXPECT_FALSE(InitializeRpcSession(&session, &segment));

  EXPECT_EQ(id, session.instance_id());
}

// TODO(chrisha): A more involved unittest where we launch a child process
//     whose RPC connection is mandatory, but unable to be initialized. Make
//     sure that it crashes as expected.

}  // namespace client
}  // namespace trace


text:
namespace stan {
namespace math {

template <typename T>
inline fvar<T> ceil(const fvar<T>& x) {
  using std::ceil;
  return fvar<T>(ceil(x.val_), 0);
}

}  // namespace math
}  // namespace stan
#endif


text:
from django.conf.urls import patterns
from django.conf.urls import url
from openstack_dashboard.dashboards.admin.policies.rules import (
    views as rule_views)
from openstack_dashboard.dashboards.admin.policies import views

POLICY = r'^(?P<policy_name>[^/]+)/%s$'

urlpatterns = patterns(
    '',
    url(r'^$', views.IndexView.as_view(), name='index'),
    url(r'^create/$', views.CreateView.as_view(), name='create'),
    url(POLICY % 'detail', views.DetailView.as_view(), name='detail'),
    url(POLICY % 'rules/create',
        rule_views.CreateView.as_view(), name='create_rule'),
)


text:
import { TagSyntaxRule } from './TagSyntaxRule';

export class SyntaxRuleSet {
	/**
	 * Creates an instance of the syntax rule set class this static method is called on.
	 */
	public static createInstance(): SyntaxRuleSet {
		return new this();
	}

	public static isSyntaxRuleSetClass(candidate: typeof SyntaxRuleSet): boolean {
		return (typeof candidate === 'function' && candidate._syntaxRuleSetBrand_ === SyntaxRuleSet._syntaxRuleSetBrand_);
	}

	public hasTagSyntaxRule(rule: TagSyntaxRule): boolean {
		return this.tagSyntaxRules.indexOf(rule) !== -1;
	}

	public getAllTagSyntaxRules(): TagSyntaxRule[] {
		return [].concat(this.tagSyntaxRules);
	}

	/**
	 * @chainable
	 */
	public addTagSyntaxRule(rule: TagSyntaxRule) {
		if (!this.hasTagSyntaxRule(rule)) {
			this.tagSyntaxRules.push(rule);
		}
		return this;
	}

	/**
	 * @chainable
	 */
	public addTagSyntaxRules(...rules: TagSyntaxRule[]) {
		rules.forEach(rule => this.addTagSyntaxRule(rule));
		return this;
	}

	private static _syntaxRuleSetBrand_ = Math.random();

	private tagSyntaxRules: TagSyntaxRule[] = [];
}


text:
/*
 *  sbm-bml.cpp
 *
 *  Implements the necessary BML interfaces
 *  for SBM's use of the BML libraries.
 */

#include <iostream>

#include "sbm_bml.hpp"
#include "mcontrol_util.h"

using namespace std;

///////////////////////////////////////////////////////////////////////////////
//  Helper Functions

namespace SmartBody {
	void vrSpeakFailed(
		const char* agent_id,
		const char* recipient,
		const char* message_id,
		const char* error_msg )
	{
		//  Let's not error on our error messages.  Be thorough.
		if( agent_id==NULL || agent_id[0]=='\0' )
			agent_id = "?";
		if( recipient==NULL || recipient[0]=='\0' )
			recipient = "?";
		if( message_id==NULL || message_id[0]=='\0' )
			message_id = "?";
		if( error_msg==NULL || error_msg[0]=='\0' )
			error_msg = "INVALID_ERROR_MESSAGE";

		ostringstream buff;
		cerr << "ERROR: vrSpeak: " << error_msg << "   (agent \"" << agent_id <<
			"\", recipient \"" << recipient << "\", message id \"" << message_id << "\")" << endl;
		buff << agent_id << " " << recipient << " " << message_id << " " << error_msg;
		mcuCBHandle::singleton().vhmsg_send( "vrSpeakFailed", buff.str().c_str() );
	}
};

///////////////////////////////////////////////////////////////////////////////
//  SmartBody::BmlProcessor

boost::shared_ptr<XercesDOMParser> SmartBody::BmlProcessor::xmlParser;

// Constructor
SmartBody::BmlProcessor::BmlProcessor()
:	_auto_print_schedule( false ),
	_log_synchpoints( false ),
	_warn_unknown_agents( true ),
	_parser( BML_X::Parser::new_instance() )
{
	if( xmlParser.get()==NULL ) {
		try {
			xmlParser = boost::shared_ptr<XercesDOMParser>( new XercesDOMParser() );

			xmlParser->setErrorHandler( new HandlerBase() );
			//ErrorHandler* errHandler = (ErrorHandler*) new HandlerBase();
			//xmlParser->setErrorHandler(errHandler);
			//xmlParser->setErrorHandler( new HandlerBase() );
		} catch( const XMLException& e ) {
			cerr << "ERROR: BML Processor:  XMLException during constructor: "<< e.getMessage() << endl;
		} catch( const std::exception& e ) {
			cerr << "ERROR: BML Processor:  std::exception during constructor: "<< e.what() << endl;
		} catch(...) {
			cerr << "ERROR: BML Processor:  UNKNOWN EXCEPTION DURING CONSTRUCTOR.     <<==================" << endl;
		}
	}
}

SmartBody::BmlProcessor::~BmlProcessor() {
	if( _parser ) {
		delete _parser;
		_parser = NULL;
	}
}

///////////////////////////////////////////////////////////////////////////////
//  Static Command and Message Hooks

//  Notify BodyPlanner of vrSpeak command/message.
int SmartBody::BmlProcessor::vrSpeak_func( srArgBuffer& args, mcuCBHandle *mcu ) {
#if USE_NEW_BML_PROCESSOR
	BmlProcessor& bp = mcu->bml_processor;

	const char *agent_id     = args.read_token();
	const char *recipient_id = args.read_token();
	const char *message_id   = args.read_token();
	char       *xml          = args.read_remainder_raw();
	//cout << "DEBUG: vrSpeak " << agentId << " " << recipientId << " " << messageId << endl;

	try {
		if( xml[0]=='\0' ) {
			vrSpeakFailed( agent_id, recipient_id, message_id, "vrSpeak message incomplete (empty XML argument)." );
			return CMD_FAILURE;
		}
		if( xml[0] == '"' ) {
			++xml;
			int len = strlen( xml );
			xml[--len] = '\0'; // shorten by one, assumed to be another double quote
		}
		if( strstr( xml, "cache-overwrite" )==xml ) {  // if xml begins with "cache-overwrite"
			xml+=15; // skip it
			int whitespace = TRUE;
			while( whitespace ) {
				switch( xml[0] ) {
					case ' ':
					case '\t':
						++xml;
					default:
						whitespace = FALSE;
				}
			}
		}

		SbmCharacter *agent = mcu->character_map.lookup( agent_id );
		if( agent==NULL ) {
			//  Agent is not managed by this SBM process
			if( bp.warn_unknown_agents() )
				cerr << "WARNING: BmlProcessor: Unknown agent \"" << agent_id << "\"." << endl;
			// Ignore
			return CMD_SUCCESS;
		}

		if( agent->scheduler_p==NULL ) {
			vrSpeakFailed( agent_id, recipient_id, message_id, "Uninitialized agent." );
			return CMD_FAILURE;
		}

        DOMDocument *xmlDoc = xml_utils::parseMessageXml( xmlParser.get(), xml );
		if( xmlDoc == NULL ) {
			vrSpeakFailed( agent_id, recipient_id, message_id, "XML parser returned NULL document." );
			return CMD_FAILURE;
		}

		//  TODO!!!
		vrSpeakFailed( agent_id, recipient_id, message_id, "vrSpeak Unimplemented." );
		return CMD_FAILURE;

		//BodyPlannerMsg bpMsg( agent_id, recipient_id, message_id, agent, xmlDoc );
		//bp.vrSpeak( bpMsg, mcu );

		//return( CMD_SUCCESS );
	//} catch( BodyPlannerException& e ) {
	//	ostringstream msg;
	//	msg << "BodyPlannerException: "<<e.message;
	//	vrSpeakFailed( agent_id, recipient_id, message_id, msg.str().c_str() );
	//	return CMD_FAILURE;
	} catch( const std::exception& e ) {
		ostringstream msg;
		msg << "std::exception: "<<e.what();
		vrSpeakFailed( agent_id, recipient_id, message_id, msg.str().c_str() );
		return CMD_FAILURE;
	//} catch( ... ) {
	//	ostringstream msg;
	//	msg << "Unknown exception."<<e.message;
	//	vrSpeakFailed( agent_id, recipient_id, message_id, msg.str().c_str() );
	//	return CMD_FAILURE;
	}
#else
	cout << "ERROR: SmartBody::BmlProcessor::vrSpeak_func(..) called when compiled without USE_NEW_BML_PROCESSOR." << endl;
	return CMD_FAILURE;
#endif  // USE_NEW_BML_PROCESSOR
}

// Notify BodyPlanner of vrSpoke messages.
int SmartBody::BmlProcessor::vrSpoke_func( srArgBuffer& args, mcuCBHandle *mcu ) {
#if USE_NEW_BML_PROCESSOR
	cerr << "ERROR: SmartBody::BmlProcessor::vrSpoke_func(..): Unimplemented." << endl;
	return CMD_FAILURE;
#else
	cout << "ERROR: SmartBody::BmlProcessor::vrSpoke_func(..) called when compiled without USE_NEW_BML_PROCESSOR." << endl;
	return CMD_FAILURE;
#endif  // USE_NEW_BML_PROCESSOR
}

// Notify BodyPlanner of completed speech request.
int SmartBody::BmlProcessor::bpSpeechReady_func( srArgBuffer& args, mcuCBHandle *mcu ) {
#if USE_NEW_BML_PROCESSOR
	cerr << "ERROR: SmartBody::BmlProcessor::bpSpeechReady_func(..): Unimplemented." << endl;
	return CMD_FAILURE;
#else
	cout << "ERROR: SmartBody::BmlProcessor::bpSpeechReady_func(..) called when compiled without USE_NEW_BML_PROCESSOR." << endl;
	return CMD_FAILURE;
#endif  // USE_NEW_BML_PROCESSOR
}

// Notify BodyPlanner of request timings.
int SmartBody::BmlProcessor::bp_cmd_func( srArgBuffer& args, mcuCBHandle *mcu ) {
#if USE_NEW_BML_PROCESSOR
	cerr << "ERROR: SmartBody::BmlProcessor::bp_cmd_func(..): Unimplemented." << endl;
	return CMD_FAILURE;
#else
	cout << "ERROR: SmartBody::BmlProcessor::bp_cmd_func(..) called when compiled without USE_NEW_BML_PROCESSOR." << endl;
	return CMD_FAILURE;
#endif  // USE_NEW_BML_PROCESSOR
}

void set_bp_help() {
	cout << "Available BmlProcessor attributes for \"set bp ...\"" << endl
		    << "\tauto_print_schedule  [on|off]" << endl
		    << "\tlog_synchpoints      [on|off]" << endl
		    << "\twarn_unknown_agents  [on|off]" << endl;
}

// Handles the command "set bodyplanner" or "set bp"
int SmartBody::BmlProcessor::set_func( srArgBuffer& args, mcuCBHandle *mcu ) {
#if USE_NEW_BML_PROCESSOR
	BmlProcessor& bp = mcu->bml_processor;

	string attribute = args.read_token();
	if( attribute=="help" || attribute=="" ) {
		set_bp_help();
		return CMD_SUCCESS;
	} else if( attribute == "auto_print_schedule" ||
	           attribute == "auto-print-schedule" ) {
		string value = args.read_token();
		if( value == "on" ) {
			bp.auto_print_schedule( true );
			return CMD_SUCCESS;
		} else if( value == "off" ) {
			bp.auto_print_schedule( false );
			return CMD_SUCCESS;
		} else {
			cerr << "ERROR: SmartBody::BmlProcessor::set_func(): expected \"on\" or \"off\" for " << attribute <<".  Found \""<<value<<"\"."<< endl;
			return CMD_FAILURE;
		}
	} else if( attribute == "log_synchpoints" ||
	           attribute == "log-synchpoints" ) {
		string value = args.read_token();
		if( value == "on" ) {
			bp.log_synchpoints( true );
			return CMD_SUCCESS;
		} else if( value == "off" ) {
			bp.log_synchpoints( false );
			return CMD_SUCCESS;
		} else {
			cerr << "ERROR: SmartBody::BmlProcessor::set_func(): expected \"on\" or \"off\" for " << attribute <<".  Found \""<<value<<"\"."<< endl;
			return CMD_FAILURE;
		}
	} else if( attribute == "warn_unknown_agents" ||
	           attribute == "warn-unknown-agents" ) {
		string value = args.read_token();
		if( value == "on" ) {
			bp.warn_unknown_agents( true );
			return CMD_SUCCESS;
		} else if( value == "off" ) {
			bp.warn_unknown_agents( false );
			return CMD_SUCCESS;
		} else {
			cerr << "ERROR: SmartBody::BmlProcessor::set_func(): expected \"on\" or \"off\" for " << attribute <<".  Found \""<<value<<"\"."<< endl;
			return CMD_FAILURE;
		}
	} else {
		cerr << "ERROR: SmartBody::BmlProcessor::set_func(): Unknown attribute \"" << attribute <<"\"."<< endl;
		set_bp_help();
        return CMD_NOT_FOUND;
	}
#else
	cout << "ERROR: SmartBody::BmlProcessor::set_func(..) called when compiled without USE_NEW_BML_PROCESSOR." << endl;
	return CMD_FAILURE;
#endif  // USE_NEW_BML_PROCESSOR
}

void print_bp_help() {
	cout << "Available BmlProcessor attributes for \"print bp ...\"" << endl
	     << "\tauto_print_schedule" << endl
	     << "\tlog_synchpoints" << endl
	     << "\twarn_unknown_agents" << endl;
}

// Handles the command "print bp"
int SmartBody::BmlProcessor::print_func( srArgBuffer& args, mcuCBHandle *mcu ) {
#if USE_NEW_BML_PROCESSOR
	BmlProcessor& bp = mcu->bml_processor;

	string attribute = args.read_token();
	if( attribute=="help" || attribute=="" ) {
		print_bp_help();
		return CMD_SUCCESS;
	} else if( attribute == "auto_print_schedule" ||
	           attribute == "auto-print-schedule" ) {
		cout << "BML Processor auto_print_schedule: "<<
			(bp.auto_print_schedule()? "on" : "off") << endl;
		return CMD_SUCCESS;
	} else if( attribute == "log_synchpoints" ||
	           attribute == "log-synchpoints" ) {
		cout << "BML Processor log_synchpoints: "<<
			(bp.log_synchpoints()? "on" : "off") << endl;
		return CMD_SUCCESS;
	} else if( attribute == "warn_unknown_agents" ||
	           attribute == "warn-unknown-agents" ) {
		cout << "BML Processor warn_unknown_agents: "<<
			(bp.warn_unknown_agents()? "on" : "off") << endl;
		return CMD_SUCCESS;
	} else {
		cerr << "ERROR: SmartBody::BmlProcessor::print_func(): Unknown attribute \"" << attribute <<"\"."<< endl;
		print_bp_help();
        return CMD_NOT_FOUND;
	}
#else
	cout << "ERROR: SmartBody::BmlProcessor::print_func(..) called when compiled without USE_NEW_BML_PROCESSOR." << endl;
	return CMD_FAILURE;
#endif  // USE_NEW_BML_PROCESSOR
}


text:
package main

import (
	"bytes"
	"encoding/xml"
	"flag"
	"fmt"
	"io/ioutil"
	"net/http"
	"os"
	"strconv"
	"time"
)

const fmtString = `// Copyright 2012 Google, Inc. All rights reserved.

package layers

// Created by gen.go, don't edit manually
// Generated at %s
// Fetched from %q

// TCPPortNames contains the port names for all TCP ports.
var TCPPortNames = tcpPortNames

// UDPPortNames contains the port names for all UDP ports.
var UDPPortNames = udpPortNames

// SCTPPortNames contains the port names for all SCTP ports.
var SCTPPortNames = sctpPortNames

var tcpPortNames = map[TCPPort]string{
%s}
var udpPortNames = map[UDPPort]string{
%s}
var sctpPortNames = map[SCTPPort]string{
%s}
`

var url = flag.String("url", "http://www.iana.org/assignments/service-names-port-numbers/service-names-port-numbers.xml", "URL to grab port numbers from")

func main() {
	fmt.Fprintf(os.Stderr, "Fetching ports from %q\n", *url)
	resp, err := http.Get(*url)
	if err != nil {
		panic(err)
	}
	defer resp.Body.Close()
	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		panic(err)
	}
	fmt.Fprintln(os.Stderr, "Parsing XML")
	var registry struct {
		Records []struct {
			Protocol string `xml:"protocol"`
			Number   string `xml:"number"`
			Name     string `xml:"name"`
		} `xml:"record"`
	}
	xml.Unmarshal(body, &registry)
	var tcpPorts bytes.Buffer
	var udpPorts bytes.Buffer
	var sctpPorts bytes.Buffer
	done := map[string]map[int]bool{
		"tcp":  {},
		"udp":  {},
		"sctp": {},
	}
	for _, r := range registry.Records {
		port, err := strconv.Atoi(r.Number)
		if err != nil {
			continue
		}
		if r.Name == "" {
			continue
		}
		var b *bytes.Buffer
		switch r.Protocol {
		case "tcp":
			b = &tcpPorts
		case "udp":
			b = &udpPorts
		case "sctp":
			b = &sctpPorts
		default:
			continue
		}
		if done[r.Protocol][port] {
			continue
		}
		done[r.Protocol][port] = true
		fmt.Fprintf(b, "\t%d: %q,\n", port, r.Name)
	}
	fmt.Fprintln(os.Stderr, "Writing results to stdout")
	fmt.Printf(fmtString, time.Now(), *url, tcpPorts.String(), udpPorts.String(), sctpPorts.String())
}


text:
package com.sequenceiq.cloudbreak.reactor.handler.cluster;

import javax.inject.Inject;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import com.sequenceiq.cloudbreak.core.cluster.ClusterUpscaleService;
import com.sequenceiq.cloudbreak.eventbus.Event;
import com.sequenceiq.cloudbreak.eventbus.EventBus;
import com.sequenceiq.cloudbreak.reactor.api.event.cluster.ClusterManagerInstallComponentsRequest;
import com.sequenceiq.cloudbreak.reactor.api.event.cluster.ClusterManagerInstallComponentsResult;
import com.sequenceiq.flow.event.EventSelectorUtil;
import com.sequenceiq.flow.reactor.api.handler.EventHandler;

@Component
public class AmbariInstallComponentsHandler implements EventHandler<ClusterManagerInstallComponentsRequest> {

    private static final Logger LOGGER = LoggerFactory.getLogger(AmbariInstallComponentsHandler.class);

    @Inject
    private EventBus eventBus;

    @Inject
    private ClusterUpscaleService clusterUpscaleService;

    @Override
    public String selector() {
        return EventSelectorUtil.selector(ClusterManagerInstallComponentsRequest.class);
    }

    @Override
    public void accept(Event<ClusterManagerInstallComponentsRequest> event) {
        ClusterManagerInstallComponentsRequest request = event.getData();
        Long stackId = request.getResourceId();
        ClusterManagerInstallComponentsResult result;
        try {
            clusterUpscaleService.installComponents(stackId, request.getComponents(), request.getHostName());
            result = new ClusterManagerInstallComponentsResult(request);
        } catch (Exception e) {
            String message = "Failed to install components on new host";
            LOGGER.error(message, e);
            result = new ClusterManagerInstallComponentsResult(message, e, request);
        }
        eventBus.notify(result.selector(), new Event<>(event.getHeaders(), result));

    }
}


text:
package main

import (
	"bytes"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"log"
	"math"
	"net"
	"net/http"
	"net/url"
	"path"
	"regexp"
	"strconv"
	"strings"

	"github.com/gorilla/mux"
	"github.com/xiam/hyperfox/tools/capture"
	"menteslibres.net/gosexy/to"
	"upper.io/db"
)

var (
	cleanPattern  = regexp.MustCompile(`[^0-9a-zA-Z\s\.]`)
	spacesPattern = regexp.MustCompile(`\s+`)
)

const (
	serviceBindHost      = `127.0.0.1`
	serviceBindStartPort = 3030
)

const (
	pageSize         = uint(50)
	directionRequest = `req`
)

type getResponse struct {
	capture.Response `json:",inline"`
}

func (g getResponse) Constraint() db.Cond {
	return db.Cond{"id": g.ID}
}

type pullResponse struct {
	Data  []capture.Response `json:"data"`
	Pages uint               `json:"pages"`
	Page  uint               `json:"page"`
}

func replyJSON(wri http.ResponseWriter, data interface{}) {
	var buf []byte
	var err error

	if buf, err = json.Marshal(data); err != nil {
		log.Printf("Marshal: %q", err)
		return
	}

	wri.Header().Set("Access-Control-Allow-Origin", "*")

	wri.WriteHeader(http.StatusOK)
	wri.Write(buf)
}

func rootHandler(wri http.ResponseWriter, req *http.Request) {

}

// downloadHandler provides a downloadable document given its ID.
func downloadHandler(wri http.ResponseWriter, req *http.Request) {

	var err error
	var response getResponse

	if err = req.ParseForm(); err != nil {
		log.Printf("ParseForm: %q", err)
		return
	}

	wireFormat := to.Bool(req.Form.Get("wire"))
	direction := req.Form.Get("type")

	response.ID = uint(to.Int64(req.Form.Get("id")))

	res := col.Find(response)

	res.Select(
		"id",
		"url",
		"method",
		"header",
		"request_header",
		"date_end",
		db.Raw{"hex(body) AS body"},
		db.Raw{"hex(request_body) AS request_body"},
	)

	if err = res.One(&response.Response); err != nil {
		log.Printf("res.One: %q", err)
		return
	}

	var u *url.URL

	if u, err = url.Parse(response.URL); err != nil {
		log.Printf("url.Parse: %q", err)
		return
	}

	var body []byte

	basename := path.Base(u.Path)

	switch direction {
	case directionRequest:
		if body, err = hex.DecodeString(string(response.RequestBody)); err != nil {
			log.Printf("url.Parse: %q", err)
			return
		}

		if wireFormat {

			buf := bytes.NewBuffer(nil)

			buf.WriteString(fmt.Sprintf("%s %s HTTP/1.1\r\n", response.Method, u.RequestURI()))

			for k, vv := range response.RequestHeader.Header {
				for _, v := range vv {
					buf.WriteString(fmt.Sprintf("%s: %s\r\n", k, v))
				}
			}

			buf.WriteString("\r\n")

			wri.Header().Set("Content-Disposition", `attachment; filename="`+u.Host+"-"+basename+`.bin"`)

			buf.Write(body)

			http.ServeContent(wri, req, "", response.DateEnd, bytes.NewReader(buf.Bytes()))

		} else {

			wri.Header().Set("Content-Disposition", `attachment; filename="`+basename+`"`)

			http.ServeContent(wri, req, basename, response.DateEnd, bytes.NewReader(body))

		}

	default:
		if body, err = hex.DecodeString(string(response.Body)); err != nil {
			log.Printf("url.Parse: %q", err)
			return
		}

		if wireFormat {

			buf := bytes.NewBuffer(nil)

			for k, vv := range response.Header.Header {
				for _, v := range vv {
					buf.WriteString(fmt.Sprintf("%s: %s\r\n", k, v))
				}
			}

			buf.WriteString("\r\n")

			wri.Header().Set("Content-Disposition", `attachment; filename="`+u.Host+"-"+basename+`.bin"`)

			buf.Write(body)

			http.ServeContent(wri, req, "", response.DateEnd, bytes.NewReader(buf.Bytes()))

		} else {

			wri.Header().Set("Content-Disposition", `attachment; filename="`+basename+`"`)

			http.ServeContent(wri, req, basename, response.DateEnd, bytes.NewReader(body))

		}

	}

	return
}

// getHandler service returns a request body.
func getHandler(wri http.ResponseWriter, req *http.Request) {
	var err error
	var response getResponse

	if err = req.ParseForm(); err != nil {
		log.Printf("ParseForm: %q", err)
		return
	}

	response.ID = uint(to.Int64(req.Form.Get("id")))

	res := col.Find(response)

	res.Select(
		"id",
		"method",
		"origin",
		"content_type",
		"content_length",
		"status",
		"host",
		"url",
		"scheme",
		"header",
		"request_header",
		"date_start",
		"date_end",
		"time_taken",
	)

	if err = res.One(&response.Response); err != nil {
		log.Printf("res.One: %q", err)
		return
	}

	replyJSON(wri, response)

	return
}

// pullHandler service serves paginated requests.
func pullHandler(wri http.ResponseWriter, req *http.Request) {
	var err error
	var response pullResponse

	if err = req.ParseForm(); err != nil {
		log.Printf("ParseForm: %q", err)
		return
	}

	q := req.Form.Get("q")

	q = cleanPattern.ReplaceAllString(q, " ")
	q = spacesPattern.ReplaceAllString(q, " ")

	response.Page = uint(to.Int64(req.Form.Get("page")))

	if response.Page < 1 {
		response.Page = 1
	}

	// Result set
	res := col.Find().Select(
		"id",
		"method",
		"origin",
		"status",
		"host",
		"path",
		"scheme",
		"url",
		"content_length",
		"content_type",
		"date_start",
		"time_taken",
	).Sort("id").Limit(pageSize).Skip(pageSize * (response.Page - 1))

	if q != "" {

		terms := strings.Split(q, " ")
		conds := make(db.Or, 0, len(terms))

		for _, term := range terms {
			conds = append(conds, db.Or{
				db.Raw{`host LIKE '%` + term + `%'`},
				db.Raw{`origin LIKE '%` + term + `%'`},
				db.Raw{`path LIKE '%` + term + `%'`},
				db.Raw{`content_type LIKE '%` + term + `%'`},
				db.Cond{"method": term},
				db.Cond{"scheme": term},
				db.Cond{"status": term},
			},
			)
		}

		res.Where(conds...)
	}

	// Pulling information page.
	if err = res.All(&response.Data); err != nil {
		log.Printf("res.All: %q", err)
		return
	}

	// Getting total number of pages.
	if c, err := res.Count(); err == nil {
		response.Pages = uint(math.Ceil(float64(c) / float64(pageSize)))
	}

	replyJSON(wri, response)

	return
}

// startServices starts an http server that provides websocket and rest
// services.
func startServices() error {

	r := mux.NewRouter()

	r.HandleFunc("/", rootHandler)
	r.HandleFunc("/pull", pullHandler)
	r.HandleFunc("/get", getHandler)
	r.HandleFunc("/download", downloadHandler)

	errc := make(chan error)

	go func(errc chan error) {

		var err error
		var addr string
		var ln net.Listener

		log.Printf("Starting (local) API server...")

		// Looking for a port to listen to.
		for i := 0; i < 65535; i++ {
			addr = serviceBindHost + ":" + strconv.Itoa(serviceBindStartPort+i)
			if ln, err = net.Listen("tcp", addr); err == nil {
				// We have a listener!
				break
			}
			if strings.Contains(err.Error(), "address already in use") == false {
				// We don't know how to handle this error.
				errc <- err
				return
			}
		}

		log.Printf("Watch live capture at http://live.hyperfox.org/#/?source=%s", addr)

		srv := &http.Server{
			Addr:    addr,
			Handler: r,
		}

		// Serving API.
		go func() {
			if err := srv.Serve(ln); err != nil {
				panic(err.Error())
			}
		}()

		errc <- nil

	}(errc)

	err := <-errc

	return err
}


text:
namespace Ocelot.IntegrationTests
{
    using Administration;
    using Configuration.File;
    using DependencyInjection;
    using Microsoft.AspNetCore.Hosting;
    using Microsoft.Data.Sqlite;
    using Microsoft.Extensions.Configuration;
    using Microsoft.Extensions.DependencyInjection;
    using Middleware;
    using Newtonsoft.Json;
    using Ocelot.Provider.Rafty;
    using Rafty.Infrastructure;
    using Shouldly;
    using System;
    using System.Collections.Generic;
    using System.IO;
    using System.Linq;
    using System.Net.Http;
    using System.Net.Http.Headers;
    using System.Threading;
    using System.Threading.Tasks;
    using Xunit;
    using Xunit.Abstractions;
    using Wait = Rafty.Infrastructure.Wait;

    public class RaftTests : IDisposable
    {
        private readonly List<IWebHost> _builders;
        private readonly List<IWebHostBuilder> _webHostBuilders;
        private readonly List<Thread> _threads;
        private FilePeers _peers;
        private HttpClient _httpClient;
        private readonly HttpClient _httpClientForAssertions;
        private BearerToken _token;
        private HttpResponseMessage _response;
        private static readonly object _lock = new object();
        private ITestOutputHelper _output;

        public RaftTests(ITestOutputHelper output)
        {
            _output = output;
            _httpClientForAssertions = new HttpClient();
            _webHostBuilders = new List<IWebHostBuilder>();
            _builders = new List<IWebHost>();
            _threads = new List<Thread>();
        }

        [Fact(Skip = "Still not stable, more work required in rafty..")]
        public async Task should_persist_command_to_five_servers()
        {
            var peers = new List<FilePeer>
            {
                new FilePeer {HostAndPort = "http://localhost:5000"},

                new FilePeer {HostAndPort = "http://localhost:5001"},

                new FilePeer {HostAndPort = "http://localhost:5002"},

                new FilePeer {HostAndPort = "http://localhost:5003"},

                new FilePeer {HostAndPort = "http://localhost:5004"}
            };

            var configuration = new FileConfiguration
            {
                GlobalConfiguration = new FileGlobalConfiguration
                {
                }
            };

            var updatedConfiguration = new FileConfiguration
            {
                GlobalConfiguration = new FileGlobalConfiguration
                {
                },
                ReRoutes = new List<FileReRoute>()
                {
                    new FileReRoute()
                    {
                        DownstreamHostAndPorts = new List<FileHostAndPort>
                        {
                            new FileHostAndPort
                            {
                                Host = "127.0.0.1",
                                Port = 80,
                            }
                        },
                        DownstreamScheme = "http",
                        DownstreamPathTemplate = "/geoffrey",
                        UpstreamHttpMethod = new List<string> { "get" },
                        UpstreamPathTemplate = "/"
                    },
                    new FileReRoute()
                    {
                        DownstreamHostAndPorts = new List<FileHostAndPort>
                        {
                            new FileHostAndPort
                            {
                                Host = "123.123.123",
                                Port = 443,
                            }
                        },
                        DownstreamScheme = "https",
                        DownstreamPathTemplate = "/blooper/{productId}",
                        UpstreamHttpMethod = new List<string> { "post" },
                        UpstreamPathTemplate = "/test"
                    }
                }
            };

            var command = new UpdateFileConfiguration(updatedConfiguration);
            GivenThePeersAre(peers);
            GivenThereIsAConfiguration(configuration);
            GivenFiveServersAreRunning();
            await GivenIHaveAnOcelotToken("/administration");
            await WhenISendACommandIntoTheCluster(command);
            Thread.Sleep(5000);
            await ThenTheCommandIsReplicatedToAllStateMachines(command);
        }

        [Fact(Skip = "Still not stable, more work required in rafty..")]
        public async Task should_persist_command_to_five_servers_when_using_administration_api()
        {
            var peers = new List<FilePeer>
            {
                new FilePeer {HostAndPort = "http://localhost:5005"},

                new FilePeer {HostAndPort = "http://localhost:5006"},

                new FilePeer {HostAndPort = "http://localhost:5007"},

                new FilePeer {HostAndPort = "http://localhost:5008"},

                new FilePeer {HostAndPort = "http://localhost:5009"}
            };

            var configuration = new FileConfiguration
            {
            };

            var updatedConfiguration = new FileConfiguration
            {
                ReRoutes = new List<FileReRoute>()
                {
                    new FileReRoute()
                    {
                        DownstreamHostAndPorts = new List<FileHostAndPort>
                        {
                            new FileHostAndPort
                            {
                                Host = "127.0.0.1",
                                Port = 80,
                            }
                        },
                        DownstreamScheme = "http",
                        DownstreamPathTemplate = "/geoffrey",
                        UpstreamHttpMethod = new List<string> { "get" },
                        UpstreamPathTemplate = "/"
                    },
                    new FileReRoute()
                    {
                        DownstreamHostAndPorts = new List<FileHostAndPort>
                        {
                            new FileHostAndPort
                            {
                                Host = "123.123.123",
                                Port = 443,
                            }
                        },
                        DownstreamScheme = "https",
                        DownstreamPathTemplate = "/blooper/{productId}",
                        UpstreamHttpMethod = new List<string> { "post" },
                        UpstreamPathTemplate = "/test"
                    }
                }
            };

            var command = new UpdateFileConfiguration(updatedConfiguration);
            GivenThePeersAre(peers);
            GivenThereIsAConfiguration(configuration);
            GivenFiveServersAreRunning();
            await GivenIHaveAnOcelotToken("/administration");
            GivenIHaveAddedATokenToMyRequest();
            await WhenIPostOnTheApiGateway("/administration/configuration", updatedConfiguration);
            await ThenTheCommandIsReplicatedToAllStateMachines(command);
        }

        private void GivenThePeersAre(List<FilePeer> peers)
        {
            FilePeers filePeers = new FilePeers();
            filePeers.Peers.AddRange(peers);
            var json = JsonConvert.SerializeObject(filePeers);
            File.WriteAllText("peers.json", json);
            _httpClient = new HttpClient();
            var ocelotBaseUrl = peers[0].HostAndPort;
            _httpClient.BaseAddress = new Uri(ocelotBaseUrl);
        }

        private async Task WhenISendACommandIntoTheCluster(UpdateFileConfiguration command)
        {
            async Task<bool> SendCommand()
            {
                try
                {
                    var p = _peers.Peers.First();
                    var json = JsonConvert.SerializeObject(command, new JsonSerializerSettings()
                    {
                        TypeNameHandling = TypeNameHandling.All
                    });
                    var httpContent = new StringContent(json);
                    httpContent.Headers.ContentType = new MediaTypeHeaderValue("application/json");
                    using (var httpClient = new HttpClient())
                    {
                        httpClient.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", _token.AccessToken);
                        var response = await httpClient.PostAsync($"{p.HostAndPort}/administration/raft/command", httpContent);
                        response.EnsureSuccessStatusCode();
                        var content = await response.Content.ReadAsStringAsync();

                        var errorResult = JsonConvert.DeserializeObject<ErrorResponse<UpdateFileConfiguration>>(content);

                        if (!string.IsNullOrEmpty(errorResult.Error))
                        {
                            return false;
                        }

                        var okResult = JsonConvert.DeserializeObject<OkResponse<UpdateFileConfiguration>>(content);

                        if (okResult.Command.Configuration.ReRoutes.Count == 2)
                        {
                            return true;
                        }
                    }

                    return false;
                }
                catch (Exception e)
                {
                    Console.WriteLine(e);
                    return false;
                }
            }

            var commandSent = await Wait.WaitFor(40000).Until(async () =>
            {
                var result = await SendCommand();
                Thread.Sleep(1000);
                return result;
            });

            commandSent.ShouldBeTrue();
        }

        private async Task ThenTheCommandIsReplicatedToAllStateMachines(UpdateFileConfiguration expecteds)
        {
            async Task<bool> CommandCalledOnAllStateMachines()
            {
                try
                {
                    var passed = 0;
                    foreach (var peer in _peers.Peers)
                    {
                        var path = $"{peer.HostAndPort.Replace("/", "").Replace(":", "")}.db";
                        using (var connection = new SqliteConnection($"Data Source={path};"))
                        {
                            connection.Open();
                            var sql = @"select count(id) from logs";
                            using (var command = new SqliteCommand(sql, connection))
                            {
                                var index = Convert.ToInt32(command.ExecuteScalar());
                                index.ShouldBe(1);
                            }
                        }

                        _httpClientForAssertions.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", _token.AccessToken);
                        var result = await _httpClientForAssertions.GetAsync($"{peer.HostAndPort}/administration/configuration");
                        var json = await result.Content.ReadAsStringAsync();
                        var response = JsonConvert.DeserializeObject<FileConfiguration>(json, new JsonSerializerSettings { TypeNameHandling = TypeNameHandling.All });
                        response.GlobalConfiguration.RequestIdKey.ShouldBe(expecteds.Configuration.GlobalConfiguration.RequestIdKey);
                        response.GlobalConfiguration.ServiceDiscoveryProvider.Host.ShouldBe(expecteds.Configuration.GlobalConfiguration.ServiceDiscoveryProvider.Host);
                        response.GlobalConfiguration.ServiceDiscoveryProvider.Port.ShouldBe(expecteds.Configuration.GlobalConfiguration.ServiceDiscoveryProvider.Port);

                        for (var i = 0; i < response.ReRoutes.Count; i++)
                        {
                            for (var j = 0; j < response.ReRoutes[i].DownstreamHostAndPorts.Count; j++)
                            {
                                var res = response.ReRoutes[i].DownstreamHostAndPorts[j];
                                var expected = expecteds.Configuration.ReRoutes[i].DownstreamHostAndPorts[j];
                                res.Host.ShouldBe(expected.Host);
                                res.Port.ShouldBe(expected.Port);
                            }

                            response.ReRoutes[i].DownstreamPathTemplate.ShouldBe(expecteds.Configuration.ReRoutes[i].DownstreamPathTemplate);
                            response.ReRoutes[i].DownstreamScheme.ShouldBe(expecteds.Configuration.ReRoutes[i].DownstreamScheme);
                            response.ReRoutes[i].UpstreamPathTemplate.ShouldBe(expecteds.Configuration.ReRoutes[i].UpstreamPathTemplate);
                            response.ReRoutes[i].UpstreamHttpMethod.ShouldBe(expecteds.Configuration.ReRoutes[i].UpstreamHttpMethod);
                        }

                        passed++;
                    }

                    return passed == 5;
                }
                catch (Exception e)
                {
                    //_output.WriteLine($"{e.Message}, {e.StackTrace}");
                    Console.WriteLine(e);
                    return false;
                }
            }

            var commandOnAllStateMachines = await Wait.WaitFor(40000).Until(async () =>
            {
                var result = await CommandCalledOnAllStateMachines();
                Thread.Sleep(1000);
                return result;
            });

            commandOnAllStateMachines.ShouldBeTrue();
        }

        private async Task WhenIPostOnTheApiGateway(string url, FileConfiguration updatedConfiguration)
        {
            async Task<bool> SendCommand()
            {
                var json = JsonConvert.SerializeObject(updatedConfiguration);

                var content = new StringContent(json);

                content.Headers.ContentType = new MediaTypeHeaderValue("application/json");

                _response = await _httpClient.PostAsync(url, content);

                var responseContent = await _response.Content.ReadAsStringAsync();

                if (responseContent == "There was a problem. This error message sucks raise an issue in GitHub.")
                {
                    return false;
                }

                if (string.IsNullOrEmpty(responseContent))
                {
                    return false;
                }

                return _response.IsSuccessStatusCode;
            }

            var commandSent = await Wait.WaitFor(40000).Until(async () =>
            {
                var result = await SendCommand();
                Thread.Sleep(1000);
                return result;
            });

            commandSent.ShouldBeTrue();
        }

        private void GivenIHaveAddedATokenToMyRequest()
        {
            _httpClient.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", _token.AccessToken);
        }

        private async Task GivenIHaveAnOcelotToken(string adminPath)
        {
            async Task<bool> AddToken()
            {
                try
                {
                    var tokenUrl = $"{adminPath}/connect/token";
                    var formData = new List<KeyValuePair<string, string>>
                    {
                        new KeyValuePair<string, string>("client_id", "admin"),
                        new KeyValuePair<string, string>("client_secret", "secret"),
                        new KeyValuePair<string, string>("scope", "admin"),
                        new KeyValuePair<string, string>("grant_type", "client_credentials")
                    };
                    var content = new FormUrlEncodedContent(formData);

                    var response = await _httpClient.PostAsync(tokenUrl, content);
                    var responseContent = await response.Content.ReadAsStringAsync();
                    if (!response.IsSuccessStatusCode)
                    {
                        return false;
                    }

                    _token = JsonConvert.DeserializeObject<BearerToken>(responseContent);
                    var configPath = $"{adminPath}/.well-known/openid-configuration";
                    response = await _httpClient.GetAsync(configPath);
                    return response.IsSuccessStatusCode;
                }
                catch (Exception)
                {
                    return false;
                }
            }

            var addToken = await Wait.WaitFor(40000).Until(async () =>
            {
                var result = await AddToken();
                Thread.Sleep(1000);
                return result;
            });

            addToken.ShouldBeTrue();
        }

        private void GivenThereIsAConfiguration(FileConfiguration fileConfiguration)
        {
            var configurationPath = $"{Directory.GetCurrentDirectory()}/ocelot.json";

            var jsonConfiguration = JsonConvert.SerializeObject(fileConfiguration);

            if (File.Exists(configurationPath))
            {
                File.Delete(configurationPath);
            }

            File.WriteAllText(configurationPath, jsonConfiguration);

            var text = File.ReadAllText(configurationPath);

            configurationPath = $"{AppContext.BaseDirectory}/ocelot.json";

            if (File.Exists(configurationPath))
            {
                File.Delete(configurationPath);
            }

            File.WriteAllText(configurationPath, jsonConfiguration);

            text = File.ReadAllText(configurationPath);
        }

        private void GivenAServerIsRunning(string url)
        {
            lock (_lock)
            {
                IWebHostBuilder webHostBuilder = new WebHostBuilder();
                webHostBuilder.UseUrls(url)
                    .UseKestrel()
                    .UseContentRoot(Directory.GetCurrentDirectory())
                    .ConfigureAppConfiguration((hostingContext, config) =>
                    {
                        config.SetBasePath(hostingContext.HostingEnvironment.ContentRootPath);
                        var env = hostingContext.HostingEnvironment;
                        config.AddJsonFile("appsettings.json", optional: true, reloadOnChange: false)
                            .AddJsonFile($"appsettings.{env.EnvironmentName}.json", optional: true, reloadOnChange: false);
                        config.AddJsonFile("ocelot.json", false, false);
                        config.AddJsonFile("peers.json", optional: true, reloadOnChange: false);
#pragma warning disable CS0618
                        config.AddOcelotBaseUrl(url);
#pragma warning restore CS0618
                        config.AddEnvironmentVariables();
                    })
                    .ConfigureServices(x =>
                    {
                        x.AddSingleton(new NodeId(url));
                        x
                            .AddOcelot()
                            .AddAdministration("/administration", "secret")
                            .AddRafty();
                    })
                    .Configure(app =>
                    {
                        app.UseOcelot().Wait();
                    });

                var builder = webHostBuilder.Build();
                builder.Start();

                _webHostBuilders.Add(webHostBuilder);
                _builders.Add(builder);
            }
        }

        private void GivenFiveServersAreRunning()
        {
            var bytes = File.ReadAllText("peers.json");
            _peers = JsonConvert.DeserializeObject<FilePeers>(bytes);

            foreach (var peer in _peers.Peers)
            {
                File.Delete(peer.HostAndPort.Replace("/", "").Replace(":", ""));
                File.Delete($"{peer.HostAndPort.Replace("/", "").Replace(":", "")}.db");
                var thread = new Thread(() => GivenAServerIsRunning(peer.HostAndPort));
                thread.Start();
                _threads.Add(thread);
            }
        }

        public void Dispose()
        {
            foreach (var builder in _builders)
            {
                builder?.Dispose();
            }

            foreach (var peer in _peers.Peers)
            {
                try
                {
                    File.Delete(peer.HostAndPort.Replace("/", "").Replace(":", ""));
                    File.Delete($"{peer.HostAndPort.Replace("/", "").Replace(":", "")}.db");
                }
                catch (Exception e)
                {
                    Console.WriteLine(e);
                }
            }
        }
    }
}