4
\$\begingroup\$

We have a set of time slices, in the order of 50. Throughout this question I will use 6. We group these slices into blocks, in this example the block size is 2. Then we want to iterate through all source slice and sink slice combination and call functions of the following types:

f(int source_slice, int sink_slice);
g(int source_slice, int sink_block);

This is how the slices are put into blocks:

enter image description here

What we can basically do is something like this:

for (int source_slice = 0; source_slice < num_slices; ++source_slice) {
  for (int sink_slice = 0; sink_slice < num_slices; ++sink_slice) {
    int const sink_block = sink_slice / block_size;
    f(source_slice, sink_slice);
    g(source_slice, sink_block);
  }
}

The data that we work with has the property that we can re-use intermediate results as long as we have the same source and sink block. Therefore this scheme wastes opportunities to re-use data. We cannot fit all the data into the RAM, so we want to iterate through this more intelligently.

This is the iteration scheme that we want:

enter image description here

We go through the blocks, and within each block combination, we take all slice combinations. The block combination with switched sink and source blocks is to be done right after because we can re-use a bit of data with that as well.

For the functions of type g, we want to iterate from slices to blocks, like this:

enter image description here

Compared to the slice-to-slice variant, we are only taking every second (every block size in general) element from the iteration.

In the current state of the code, this is done with a lot of nested for loops and a lot of repeated index calculations. Despite being repetition of code and mixing of different tasks in the same function, it is also inflexible. We want to support a second scheme, the interlace one:

enter image description here

The slice-to-slice iteration looks as follows:

enter image description here

This change would require rewriting the index fiddling with other index fiddling. Instead, I want the client code to be really simple, even if that gives me headaches in other parts of the code.

There are two factions in the notations, one will call this scheme “block 3” because it has three blocks, other call it “block 2” because the block size is two. That is represented in the code by having it parameterized by num_block and block_size in various parts.

This is what I want the client code to look like:

enum class DilutionType { block, interlace };

inline void test_dilution_scheme(int const num_slice,
                                 int const num_block,
                                 DilutionType const type) {
  auto const name = dilution_names[type];
  auto const block_size = num_slice / num_block;
  std::cout << "T = " << num_slice << ", T" << name << num_block << " (Morningstar), T"
            << name << block_size << " (Bonn):\n\n";

  DilutionScheme dilution_scheme(num_slice, block_size, type);
  for (int b = 0; b < dilution_scheme.size(); ++b) {
    auto const blocks = dilution_scheme[b];
    std::cout << std::setw(2) << blocks.source() << " => " << std::setw(2)
              << blocks.sink() << "\n";

    for (auto const slices : blocks) {
      std::cout << "  " << std::setw(2) << slices.source() << " -> " << std::setw(2)
                << slices.sink() << "\n";
    }
  }

  std::cout << "\n\n";
}

The outer loop does not use range-based-for because we want to parallelize that part with OpenMP and that needs to see the range explicitly.

With this, we can just change the value of the enum in the DilutionScheme constructor and get a different iteration pattern. Also it allows me to iterate other the slice combinations slices in the current block combination blocks.

The output would be the following for our example:

T = 6, TB3 (Morningstar), TB2 (Other):

 0 =>  0
   0 ->  0
   1 ->  0
 0 =>  1
   0 ->  2
   1 ->  2
   2 ->  0
   3 ->  0
 0 =>  2
   0 ->  4
   1 ->  4
   4 ->  0
   5 ->  0
 1 =>  1
   2 ->  2
   3 ->  2
 1 =>  2
   2 ->  4
   3 ->  4
   4 ->  2
   5 ->  2
 2 =>  2
   4 ->  4
   5 ->  4


T = 6, TI3 (Morningstar), TI2 (Other):

 0 =>  0
   0 ->  0
   2 ->  0
   4 ->  0
 0 =>  1
   0 ->  1
   2 ->  1
   4 ->  1
   1 ->  0
   3 ->  0
   5 ->  0
 0 =>  2
   0 ->  2
   2 ->  2
   4 ->  2
   2 ->  0
   4 ->  0
 1 =>  1
   1 ->  1
   3 ->  1
   5 ->  1
 1 =>  2
   1 ->  2
   3 ->  2
   5 ->  2
   2 ->  1
   4 ->  1
 2 =>  2
   2 ->  2
   4 ->  2

For the slice-to-block iteration, one calls it like this:

for (auto const slices : blocks.one_sink_slice()) {
  // ...
}

On this side, I am quite happy. I can just iterate through the slices with the given block or interlace scheme. On the other hand, I have 298 lines of code which does this:

#include <cassert>
#include <iomanip>
#include <iostream>
#include <map>
#include <stdexcept>
#include <type_traits>

enum class DilutionType { block, interlace };

std::map<DilutionType const, std::string const> dilution_names = {
    {DilutionType::block, "B"}, {DilutionType::interlace, "I"}};

class BlockIterator {
public:
 BlockIterator(int const slice_source,
               int const slice_sink,
               int const block_source,
               int const block_sink,
               int const num_slice,
               int const num_block,
               int const pass,
               DilutionType const type,
               bool const one_sink_slice = false)
     : slice_source_(slice_source),
       slice_sink_(slice_sink),
       block_source_(block_source),
       block_sink_(block_sink),
       num_slice_(num_slice),
       num_block_(num_block),
       pass_(pass),
       type_(type),
       one_sink_slice_(one_sink_slice) {}

 BlockIterator operator*() const { return *this; }

 BlockIterator operator++() {
   if (type_ == DilutionType::block) {
     int const block_size = num_slice_ / num_block_;
     int const block_source_begin = block_source_ * block_size;
     int const block_source_end = (block_source_ + 1) * block_size;
     int const block_sink_begin = block_sink_ * block_size;
     int const block_sink_end = (block_sink_ + 1) * block_size;

     // Go to the next sink slice.
     if (one_sink_slice_) {
       slice_sink_ += block_size;
     } else {
       ++slice_sink_;
     }

     // If we iterated through all sinks in this block, we need to go to the
     // next source.
     if (slice_sink_ == block_sink_end) {
       slice_sink_ = block_sink_begin;
       ++slice_source_;
     }

     // If we iterated through all the sources in our block, we need to
     // exchange
     // source and sink block and repeat.
     if (slice_source_ == block_source_end) {
       slice_source_ = block_sink_begin;
       slice_sink_ = block_source_begin;
       ++pass_;

       if (block_source_ == block_sink_) {
         ++pass_;
       }

       std::swap(block_source_, block_sink_);
     }
   } else if (type_ == DilutionType::interlace) {
     int const block_size = num_slice_ / num_block_;
     int const block_source_begin = block_source_;
     int const block_sink_begin = block_sink_;

     // Go to the next sink slice.
     if (one_sink_slice_) {
       slice_sink_ += num_slice_;
     }
     else {
       slice_sink_ += block_size;
     }

     // If we iterated through all sinks in this block, we need to go to the
     // next source.
     if (slice_sink_ >= num_slice_) {
       slice_sink_ = block_sink_begin;
       slice_source_ += block_size;
     }

     // If we iterated through all the sources in our block, we need to
     // exchange
     // source and sink block and repeat.
     if (slice_source_ >= num_slice_) {
       slice_source_ = block_sink_begin;
       slice_sink_ = block_source_begin;
       ++pass_;

       if (block_source_ == block_sink_) {
         ++pass_;
       }

       std::swap(block_source_, block_sink_);
     }
   } else {
     throw std::domain_error("This dilution scheme is not implemented.");
   }

   return *this;
  }

  bool operator!=(BlockIterator const &other) const {
    return block_source_ != other.block_source_ ||
           block_sink_ != other.block_sink_ ||
           num_slice_ != other.num_slice_ || num_block_ != other.num_block_ ||
           pass_ != other.pass_ || type_ != other.type_;
  }

  int source() const { return slice_source_; }

  int sink() const { return slice_sink_; }

private:
  int slice_source_;
  int slice_sink_;
  int block_source_;
  int block_sink_;
  int num_slice_;
  int num_block_;
  int pass_;
  DilutionType type_;
  bool one_sink_slice_;
};

class DilutionIterator {
public:
  struct Element {
    int source;
    int sink;
  };

  DilutionIterator(int const block_source,
                   int const block_sink,
                   int const num_slice,
                   int const num_block,
                   DilutionType const type)
      : block_source_(block_source),
        block_sink_(block_sink),
        num_slice_(num_slice),
        num_block_(num_block),
        type_(type) {}

  DilutionIterator operator*() const { return *this; }

  DilutionIterator operator++() {
    ++block_sink_;

    if (block_sink_ == num_block_) {
      ++block_source_;
      block_sink_ = block_source_;
    }

    return *this;
  }

  bool operator!=(DilutionIterator const &other) const {
    return block_source_ != other.block_source_ ||
           block_sink_ != other.block_sink_ ||
           num_slice_ != other.num_slice_ || num_block_ != other.num_block_;
  }

  BlockIterator begin() const {
    int const block_size = num_slice_ / num_block_;

    if (type_ == DilutionType::block) {
      return BlockIterator(block_source_ * block_size,
                           block_sink_ * block_size,
                           block_source_,
                           block_sink_,
                           num_slice_,
                           num_block_,
                           0,
                           type_,
                           one_sink_slice_);
    } else {
      return BlockIterator(block_source_,
                           block_sink_,
                           block_source_,
                           block_sink_,
                           num_slice_,
                           num_block_,
                           0,
                           type_,
                           one_sink_slice_);
    }
  }

  BlockIterator end() const {
    int const block_size = num_slice_ / num_block_;

    if (type_ == DilutionType::block) {
      return BlockIterator(block_source_ * block_size,
                           block_sink_ * block_size,
                           block_source_,
                           block_sink_,
                           num_slice_,
                           num_block_,
                           2,
                           type_,
                           one_sink_slice_);
    } else {
      return BlockIterator(block_source_,
                           block_sink_,
                           block_source_,
                           block_sink_,
                           num_slice_,
                           num_block_,
                           2,
                           type_,
                           one_sink_slice_);
    }
  }

  int source() const { return block_source_; }

  int sink() const { return block_sink_; }

  DilutionIterator one_sink_slice() const {
      DilutionIterator copy = *this;
      copy.one_sink_slice_ = true;
      return copy;
  }

private:
  int block_source_;
  int block_sink_;
  int num_slice_;
  int num_block_;
  DilutionType type_;
  bool one_sink_slice_;
};

class DilutionScheme {
public:
  static DilutionScheme make_full_dilution(int const num_slice) {
    return DilutionScheme{num_slice, num_slice, DilutionType::block};
  }

  DilutionScheme(int const num_slice,
                 int const block_size,
                 DilutionType const type)
      : num_slice_(num_slice),
        num_block_(num_slice / block_size),
        type_(type) {}

  DilutionIterator operator[](int const i) const {
    int block_sink = 0;
    int block_source = 0;
    for (int j = 0; j < i; ++j) {
      ++block_sink;

      if (block_sink == num_block_) {
        ++block_source;
        block_sink = block_source;
      }
    }

    return DilutionIterator(
        block_source, block_sink, num_slice_, num_block_, type_);
  }

  int size() const { return num_block_ * (num_block_ + 1) / 2; }

  DilutionIterator begin() const {
    return DilutionIterator(0, 0, num_slice_, num_block_, type_);
  }

  DilutionIterator end() const {
    return DilutionIterator(
        num_block_, num_block_, num_slice_, num_block_, type_);
  }

 private:
  int num_slice_;
  int num_block_;
  DilutionType type_;
};

std::ostream &operator<<(std::ostream &os, DilutionIterator const &di) {
  os << "DilutionIterator(source=" << di.source() << ", sink=" << di.sink() << ")";
  return os;
}

std::ostream &operator<<(std::ostream &os, BlockIterator const &bi) {
  os << "BlockIterator(source=" << bi.source() << ", sink=" << bi.sink() << ")";
  return os;
}

inline void test_dilution_scheme(int const num_slice,
                                 int const num_block,
                                 DilutionType const type) {
  auto const name = dilution_names[type];
  auto const block_size = num_slice / num_block;
  std::cout << "T = " << num_slice << ", T" << name << num_block << " (Morningstar), T"
            << name << block_size << " (Bonn):\n\n";

  DilutionScheme dilution_scheme(num_slice, block_size, type);
  for (int b = 0; b < dilution_scheme.size(); ++b) {
    auto const blocks = dilution_scheme[b];
    std::cout << std::setw(2) << blocks.source() << " => " << std::setw(2)
              << blocks.sink() << "\n";

    for (auto const slices : blocks) {
      std::cout << "  " << std::setw(2) << slices.source() << " -> " << std::setw(2)
                << slices.sink() << "\n";
    }
  }

  std::cout << "\n\n";
}

There are a few things that annoy me with this code:

  • It is just so much code for something appearing much simpler. The actual logic is in the two operator++. Everything else just looks like boilerplate code. My colleagues like the simple client code, but they have a bit of trouble to get their head around these “containers” and iterators that I built.

  • I have to explicitly think about the end() state of my “container”. With Python iterators, I can just implement __next__ and then do raise StopIteration when I am done. From what I know about coroutines, these would allow me to write Python style iterators. But we are limited do C++11, so that's not an option.

  • The BlockIterator operator*() const { return *this; } is a convenience because I don't want to create yet another data structure which has a source() and sink() member. The iterator already has all the needed member variables, so I just put these getters there. By returning a copy, I don't have a problem with somebody accidentally changing the iterator.

  • The block and interlace schemes are implemented with if/else if in various places. This looks like a violation of the Open Closed Principle (OCP). I had a previous implementation where the DilutionType was a template parameter, but this made it harder to select the desired type at runtime. Also we are pretty certain that there is no third scheme.

  • The one_sink_slice() has been fudged in yesterday. It looks rather strange to have yet another bool there which does more branching in every operator++. Perhaps it would be better to just have two nested for loops for source slices and then sink slices. If one want to do slice-to-block, the last for loop is just omitted. I am not completely sure, but this might require yet another layer of iterators, and I was trying to avoid that yesterday.

The code must be C++11 and compatible with GCC and the Intel C++ Compiler.

Do you have suggestions to achieve the same results but with less dense code?

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.