<commit_before># HiTToP Concurrency Library

Contains modern C++ concurrency primitives designed for use in architectures
making heavy use of asynchronous I/O, processing.

## OrderedActionPair, OrderedActionSequence

These primitives solve the problem where one or more actions may be ready to
perform in an unspecified order, but are must be performed in a specific order.
For example, perhaps a number of results are being calculated concurrently but
they need to be inserted into an output queue in a particular order.

For example:

```C++
// The calculation we want to perform
OutputType calculate(InputType);

// The inputs to our calculations:
std::vector<Input> inputs = GenerateInputs();

// The outputs from our calculations:
std::vector<Output> outputs;
outputs.reserve(inputs.size());

// An OrderedActionSequence that controls the insertion order of outputs.
hittop::concurrent::OrderedActionSequence sequence;

// Kick off all calculations using a thread pool.
for (const auto& in : inputs) {
  auto next_action = [in, &outputs]() {
    outputs.emplace_back(calculate(in));
  };

  auto ordered_action = sequence.WrapNext(next_action);

  thread_pool.Schedule(ordered_action);
}

// As the final action in the sequence, signal a future.
std::promise<void> p;
auto signal_done = sequence.WrapAction([&p]() { p.set_value(); });
signal_done();

std::future<void> done = p.get_future();
done.wait();

// At this point, the output vector is full.
```

This can be much more efficient than more general synchronization primitives
because `OrderedActionPair` and `OrderedActionSequence` use a lock-free,
wait-free algorithm to ensure proper ordering.  The rough idea is:

```
class OrderedActionPair {
  enum States {
    initial,
    run_first_calls_both,
    run_second_calls_g
  };

  atomic<State> state = initial;
  Action g;

  RunFirst(f) {
    f();
    observed_state = state.load();
    while (true) {
      if (observed_state == run_first_calls_both) {
        return g();
      } else {
        if (state.compare_exchange(observed_state, run_second_calls_g)) {
          return;
        }
      }
    }
  }

  RunSecond(g) {
    observed_state = state.load();
    this->g = g;
    while (true) {
      if (observed_state == run_second_calls_g) {
        return g();
      } else {<commit_msg>syntax highlighting<commit_after># HiTToP Concurrency Library

Contains modern C++ concurrency primitives designed for use in architectures
making heavy use of asynchronous I/O, processing.

## OrderedActionPair, OrderedActionSequence

These primitives solve the problem where one or more actions may be ready to
perform in an unspecified order, but are must be performed in a specific order.
For example, perhaps a number of results are being calculated concurrently but
they need to be inserted into an output queue in a particular order.

For example:

```C++
// The calculation we want to perform
OutputType calculate(InputType);

// The inputs to our calculations:
std::vector<Input> inputs = GenerateInputs();

// The outputs from our calculations:
std::vector<Output> outputs;
outputs.reserve(inputs.size());

// An OrderedActionSequence that controls the insertion order of outputs.
hittop::concurrent::OrderedActionSequence sequence;

// Kick off all calculations using a thread pool.
for (const auto& in : inputs) {
  auto next_action = [in, &outputs]() {
    outputs.emplace_back(calculate(in));
  };

  auto ordered_action = sequence.WrapNext(next_action);

  thread_pool.Schedule(ordered_action);
}

// As the final action in the sequence, signal a future.
std::promise<void> p;
auto signal_done = sequence.WrapAction([&p]() { p.set_value(); });
signal_done();

std::future<void> done = p.get_future();
done.wait();

// At this point, the output vector is full.
```

This can be much more efficient than more general synchronization primitives
because `OrderedActionPair` and `OrderedActionSequence` use a lock-free,
wait-free algorithm to ensure proper ordering.  The rough idea is:

```C++
class OrderedActionPair {
  enum States {
    initial,
    run_first_calls_both,
    run_second_calls_g
  };

  atomic<State> state = initial;
  Action g;

  RunFirst(f) {
    f();
    observed_state = state.load();
    while (true) {
      if (observed_state == run_first_calls_both) {
        return g();
      } else {
        if (state.compare_exchange(observed_state, run_second_calls_g)) {
          return;
        }
      }
    }
  }

  RunSecond(g) {
    observed_state = state.load();
    this->g = g;
    while (true) {
      if (observed_state == run_second_calls_g) {
        return g();
      } else {
{'fraction_non_alphanumeric': 0.09846685381127186, 'fraction_numerical': 0.0, 'mean_word_length': 3.865546218487395, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '635625', 'n_tokens_mistral': 1372, 'n_tokens_neox': 1331, 'n_words': 470}