CRTP Pattern with Most Derived Rebind, streambuf as Example

1 The Name

CRTP(Curiously Recurring Template Pattern) is a C++ idiom in which a class X derives from a class template instantiation using X itself as template argument.

CRTP is used for static polymorphism, to avoid virtual functions, and give the compiler a chance to do the best optimization. Call of virtual function doesn't cost much, esp. with correctly predicted branch. but virtual function stop the inline optimization, and cost very much for floating point numeric algorithms.

CRTP is used for equipping object with interfaces based on a very smaller set of interfaces.

2 streambuf Example

std::streambuf is a suitable example. Most of the public/protected functions is implemented based on virtual functions overflow, underflow, uflow. Functions like xsgetn and xsputn with default implementation, based on overflow, underflow, uflow. The pointer gptr, pptr are null if it's not modified by the derived object. Every time in the loop, xsgetn / xsputn check the pointer range. If it's not empty(the pointer range modified by the derived object), memcpy can be used for performance. If it's empty(or all exhausted), uflow / overflow is called.

In some instances(char oriented), gptr and pptr are always null, we can override the implementation without the check. And in some instances, the whole buffer is contiguous, then overflow / uflow always return eof. We can override the virtual functions with these conditions. But not the non-virtual one.

CRTP is suit to solve these two problems, without dynamic cost.

template <class Derived, class Char, class Traits>
struct streambuf_base {
  // ....

  Derived & get_derived() {
    return *this;
  }

  const Derived & get_derived() const {
    return *this;
  }

  int_type snextc() {
    int_type r = traits_type::eof();
    if (get_derived().sbumpc() != traits_type::eof())
      r = get_derived().sgetc();
    return r;
  }

  // ....
};

The default snextc implementation uses sbumpc and sgetc, with get_dervied prefix, the functions are resolved from derived to template base. Only if the derived doesn't override these functions, the template base's default are taken.

For example, stringbuf, alike to std::stringbuf, can implement snextc, sbumpc and sgetc, just with one pointer comparation. All other default implementations will use this new-implemented version, thanks to get_derived prefix.

3 The Problem

The above is fine with just static polymorphism. The problems raise if we want the streambuf_base be compatible with std::streambuf, to reuse all routines who only accepts std::streambuf. So we must inherit from std::streambuf.

We can't inherit from std::streambuf in derived. If we do, any functions not re-implemented by derived will be ambiguous.

Inherit from std::streambuf in streambuf_base is fine. But then the default implementations in streambuf_base will override all in std::streambuf. It's ok in these case, but not acceptable if we inherit from another base, which provides more resonable implementations than streambuf_base.

So, what we want is an order of functions resolution. from derived to ancestor(std::streambuf in former case), and at last streambuf_base.

4 The Solution

First, we must modify the streambuf_base to take another Ancestor type parameters, and provides dummy as default.

We can check whether Ancestor is dummy in every streambuf_base's function. If Ancestor is not dummy, we just call Ancestor's version. Otherwise, the default implementation is taken.

template <class Derived, class Char, class Traits, class Ancestor=dummy>
struct streambuf_base : Ancestor {
  static const bool is_dummy = std::is_same<Ancestor, dummy>::value;
  typedef typename std::conditional<
    is_dummy, streambuf_base, Ancestor>::type ancestor_type;

  int_type snextc() {
    if (!is_dummy) return ancestor_type::snextc();
    int_type r = traits_type::eof();
    if (get_derived().sbumpc() != traits_type::eof())
      r = get_derived().sgetc();
    return r;
  }
};

This introduces some other problems. If Ancestor is another one derived from streambuf_base, the Ancestor::get_derived is wrong. So if Ancestor::snextcis taken, Ancestor::snextc will call another sbumpc and sgetc, not the most-derived object's one.

One solution to this problem is using virtual functions. But make all functions be virutal is not a good solution.

Another solution is to redefine the get_derived. Add another MostDerived paramter to streambuf_base, and add most_derived_rebind meta functor. For every Ancestor, we check whether Ancestor::most_derived_rebind exists. If it exists, streambuf_base inherited from Ancestor::most_derived_rebind::template apply::type instead of Ancestor.

template <class Derived,
          class Ancestor,
          bool=has_nested_type_most_derived_rebind<Ancestor>::value>
struct streambuf_base_rebind_ancestor {
  typedef Base type;
};

template <class Derived, class Base>
struct streambuf_base_rebind_ancestor<Derived, Ancestor, true> {
  typedef typename Ancestor::most_derived_rebind::template apply<Derived>::type type;
};

template <class Derived, class Char, class Traits,
  class Ancestor=dummy,
  class MostDerived=Derived>
struct streambuf_base : streambuf_base_rebind_ancestor<MostDerived, Ancestor>::type {
  typedef typename streambuf_base_rebind_ancestor<
    MostDerived, Ancestor>::type _ancestor_type;
  typedef typename std::conditional<
    std::is_same<_ancestor_type, dummy>::value,
    streambuf_base, _ancestor_type>::type ancestor_type;

  MostDerived & get_derived() {
    return *this;
  }

  const MostDerived & get_derived() const {
    return *this;
  }

  struct most_derived_rebind {
    template <class T>
    struct apply {
      typedef streambuf_base<Derived, Char, Traits, Ancestor, T> type;
    };
  };
};

5 The Result

With above streambuf_base, we can implements text format input/output routines in very higher performance, even faster than strtod / strtol in experiments(locale is ignored).

The floating pointer converion is done by double-conversion, in shorted mode. For printf/scanf and stringstream, the precision is 20.

The random sequence for floating is generated by random() / M_PI and for long, just random().

For fscanf and fprintf, the FILE* is obtained by fmemopen and open_memstream.

Random generated input speed(in nano-seconds)

implements / type	long	float	double
streambuf_base	45.13806234741213	87.06188110351565	193.94798608398446
strtox	56.38690736389162	99.91599140930177	246.5969282989502
fscanf	201.75892387390138	244.81490730285645	437.06397615051276
istringstream	168.8839549407959	382.54306460571297	674.1460767364503

Random generated output speed(in nano-seconds)

implements / type	long	float	double
streambuf_base	109.96006123352053	221.0190715637207	285.9951219940186
snprintf	290.8328487548828	1263.6021590270996	1915.7261646881104
fprintf	279.30502569580074	1298.2731170043944	2073.194082229614
ostringstream	122.75186352539063	1086.216874282837	1651.6398792114258

simple/naive implementation of continuation/coroutine/generator on GNU/Linux

Table of Contents

• 1 continuation
• 2 generator
• 3 coroutine
• 4 usage and test
  • 4.1 primes
  • 4.2 producer and consumer

1 continuation

In some senses, ucontext.h is already a implementation of continuation, which is available on most modern OS(standarded by POSIX). In C99, we also have setjmp/longjmp to implement continuation. But setjmp/longjmp is weaker than ucontext.h.
Folowing code are just a demo, not suit for production usage. g++-4.7 is required.

#include 
#include  // for type erasure, or, we have to make continuation a template

#define CLOG(...) // we don't have/need CLOG

struct continuation {
  template <class Func>
  continuation(const Func & func, size_t stack_size=1<<18) {
    m_stack.resize(stack_size);
    m_func = func;
    int r = getcontext(&m_ctxs[1]);
    assert(r == 0);
    m_ctxs[1].uc_link = &m_ctxs[0];
    m_ctxs[1].uc_stack.ss_sp = &m_stack[0];
    m_ctxs[1].uc_stack.ss_size = m_stack.size();
    // TODO: x86_64 is safe to pass pointer with glibc, but for other platform ...
    makecontext(&m_ctxs[1], (void (*)(void))&s_context_func, 1, this);
  }

  continuation(const continuation &r) = delete;
  continuation & operator=(const continuation &r) = delete;

  int swap() {
    if (m_in_done) return -1;
    CLOG(DEBUG, "swap from", (m_is_in ? "in" : "out"));
    m_is_in = !m_is_in;
    swapcontext(&m_ctxs[!m_is_in], &m_ctxs[m_is_in]);
    CLOG(DEBUG, "after swap");
    if (m_in_done) return 0;
    return 1;
  }

  static void s_context_func(continuation *self) {
    return self->context_func();
  }

  void context_func() {
    m_func();
    m_in_done = 1;
  }

  ucontext_t m_ctxs[2];
  std::function<void(void)> m_func;
  std::vector<char> m_stack;
  bool m_is_in = false;
  int m_in_done = 0;
};

2 generator

template <class T>
struct generator : continuation {
  template <class Func>
  generator(const Func &func)
  : continuation(func) {

  }

  void yield(const T &value) {
    m_value = value;
    swap();
  }

  T yield() {
    if (swap() <= 0) throw std::runtime_error("end of iteration");
    return m_value;
  }

  T volatile m_value;
};

3 coroutine

struct coroutine : continuation {
  template <class Func>
  generator(const Func &func)
  : continuation(func) {

  }

  void yield(coroutine &other) {
    other.swap();
  }
};

4 usage and test

4.1 primes

A simple program to print primes(not all).

bool is_prime(int n) {
  if (n % 2 == 0) return n == 2;
  for (int p = 3; p * p <= n; p += 2) {
    if (n % p == 0) return false;
  }
  return true;
}

void primes(generator<int> &g) {
  g.yield(2);
  int i = 3;
  while (1) {
    if (is_prime(i)) g.yield(i);
    i += 2;
  }
}

int main() {
  generator<int> g([&]() {
      primes(g);
    });
  while (1) {
    int p = g.yield();
    std::cout << p << std::endl;
  }
}

4.2 producer and consumer

void producer(int n, int *x, continuation & cont) {
  for (int i = 0; i < n; ++i) {
    *x = i * i;
    cont.swap();
  }
}

void consumer(int volatile *x, continuation &cont) {
  while (cont.swap() > 0) {
    CLOG(INFO, "got", *x);
  }
}

向量和矩阵的表达式模板

不是我折腾矩阵和向量, 是矩阵和向量折腾我.
早期为速度, 使用 cblas 比较多. cblas 非常不好用, double, float 函数各一 还好说, 关键是大部分函数的参数太多了, 如

void cblas_dgemm(const enum CBLAS_ORDER Order, const enum CBLAS_TRANSPOSE TransA,
                 const enum CBLAS_TRANSPOSE TransB, const int M, const int N,
                 const int K, const double alpha, const double *A,
                 const int lda, const double *B, const int ldb,
                 const double beta, double *C, const int ldc);

基本上, 每次使用, 我都测试一下 M, N, K 几个值差得比较大的情形, 看看有没 有段错误, 这个方法比认真检查参数还要管用.
后来, 用 C++ 比较多, 仿照 STL, 希望尽量泛化, 不要 double, float 各写一 遍. 还希望进一步简化, 对所有矩阵, dense 的也好, csr, csc 也好, 我 们应该只需要写 gemm(A, B).
还要将计算和存储分开, 比如 gemm(A, B) 只表示一个表达式, 并不做任何实 际运算, 需要时, 可以将结果并行地存到另一个区域. 这样, 比如写 C=gemm(A, B) * alpha + beta * C 就比上面的 cblas 简明易懂多了. 还可以 用 C.parallel_assign(gemm(A, B) * alpha + beta * C) 表示并行地计算
$$C = \alpha A B + \beta C.$$
要想实现上面的表达式, 我们需要一个 proxy 抽象, vector_proxy, matrix_proxy. 矩阵, 向量的所有公有操作, 在 proxy 里实现. 具体的向量, 矩阵继承此 proxy 即可.
对所有 vector_proxy, 定义

template <class Derived, class Value, class Reference>
struct vector_proxy {
...
size_t size() const;
value_type get(size_t idx) const;
reference_type get(size_t idx);
...
};

其中, Derived 为具体的向量, 需要定义 get, size, 并继承 vector_proxy<Derived>.
再为 vector_proxy 定义 加, 减 等基本运算, 结果也为一个 proxy, 此 proxy 的 get(size_t idx) 函数实现计算.
从 proxy 到 proxy 的赋值如下.

template <class Derived, class Value, class Reference>
struct vector_proxy {
...
 
template <class R>
void assign(const vector_proxy<R> &r) {
  size_t n = size();
  assert(r.size() == n);
  for (size_t i = 0; i < n; ++i) {
     get(i) = r.get(i);
  }
}
 
template <class R>
void parallel_assign(const vector_proxy<R> &r) {
  size_t n = size();
  assert(r.size() == n);
#pragma omp parallel for 
  for (size_t i = 0; i < n; ++i) {
    get(i) = r.get(i);
  }
}
 
...
 
}

matrix_proxy 和 vector_proxy 完全类似.
具体到 gemm_proxy 上, 如下实现 get 方法即可.

value_type get(size_t i, size_t j) {
  return vec_inner_prod(m_A.row(i), m_B.col(j));
}

矩阵的 row, col 返回的也是 vector_proxy. 在 vec_inner_prod 里, 根据向量是 dense 还是 sparse, 采用不同的计算方法即可.
gemm_proxy 里的 row(i), col(j) 可以采用矩阵和向量的乘积 gemv(m_B, m_A.row(i)), gemv(m_A, m_B.col(j)). 在 gemv_proxy 里, 还是用 vec_inner_prod 来自动区分 dense/sparse 的向量.
gemm(A, B) 在 A, B 都是稀疏矩阵时, 情况比较复杂, 使 gemv 返回一个稀 疏向量的实现效率比较低, 我直接将结果当成 dense 了.

MPI 上基于 RMA 的并行互斥锁和读写锁

MPI2 支持简单的 RMA 操作, 尤其是 Passive 模式, 取数据的进程不需要目标进 程干预, 在支持互斥锁的 Passive 模式下, 可以实现一个并行全局计数器. 这在 Using MPI 2 ¹ 里, 有比较详细的实现. 书中还提到基于此计数器实现分布互斥锁, 并给出了代码.

void MPE_Mutex_lock_simple(MPI_Win win) {
  int value;
  MPE_Counter_inc_simple(win, &value, 1);
  while (value != 0) {
    MPE_Counter_inc_simple(win, &value, -1);
    MPE_Counter_inc_simple(win, &value, 1);
  }
}

void MPE_Mutex_unlock_simple(MPI_Win win) {
  int value;
  MPE_Counter_inc_simple(win, &value, -1);
}

以上代码有严重的效率问题². 在进程数多了以后, 计数器的期望在进程数一半 处,只有非常小的概率在零附近, 从而使各进程不断的在 while 循环里. 简单 的加一个计数器是否大于零的测试再尝试加一, 减一的循环应该可以减少在 =while=循环里的进程数, 从而使计数器的值有较大概率在零附近.

class comm_mutex {
 public:
  comm_mutex(Comm &comm)
      : m_counter(comm) {
  }

  void lock() {
    while (m_counter.add(1) > 1) {
      if (m_counter.add(-1) > 0) {
        while (m_counter.get() > 0);
      }
    }
  }

  void unlock() {
    m_counter.add(-1);
  }

 private:
  comm_counter m_counter;
};

分析两种实现的效率, 即 $n$ 个进程同时加锁, 马上释放, 给出计数器 add, get 的效率后, 估算两种实现从第一个进程进锁, 到最后一个进程离锁的时间. 为简化, 我们假设各 add,get 的时间是固定的, 不随计数器的访问进程数 变化, 且进一步假设 get 的时间和 add 一样. 我们还假设计数器的互斥访 问是随机的. 最后, 我们需要看, 到最后一个进程释放锁时, 访问计数器的总次 数.
在第一个片段里, 标记程序在 unlock 里减一前为状态 L, 减一后, 为状态 R. 在 lock 里, 加一前为状态 A, 减一前为状态 S. 分别以 $l, r, a, s$ 记 某一时刻在某一状态的程序数目. 计数器的值为 $l + s$, 且 $l$ 只能为 0 或 1. 可以容易得到程序集从状态 $(l, r, a, s)$ 进一次计数器变换到另一个 状态的概率. 如从 $(l, r, a, s)$ 到 $(l, r, a + 1, s - 1)$ 的概率为 $\frac{s}{l + a + s}$.
问题即是, 程序集从状态 $(0, 0, n, 0)$ 到状态 $(0, n, 0, 0)$ 的转移次数 的期望. 有转移概率, 即可在程序上求此数值, 但表达形式应该比较复杂, 不知 道极限有没有简单的形式. 从 $s$ 的期望近似为 $\frac{n - r}{2}$ 看, 转移 次数期望有关于 $n$ 的指数下限. 实测时, 在进程数到 $32$ 后, 普遍会在释放 一两个锁后, 僵锁在加一, 减一的循环里.
第二个片段类似. 标记程序在 unlock 里减一前为状态 L, 减一后, 为状态 R. 在 lock 里, 加一前为状态 A, 减一前为状态 S, get 前为状态 G, 分别 以 $l, r, a, s, g$ 记某一时刻在某一状态的程序数目. 问题变成程序集从状态 $(0, 0, n, 0, 0)$ 到状态 $(0, n, 0, 0, 0)$ 的转移次数的期望.
$s$ 大后, 多数程序会转移到状态 G, 从而 $s$ 的期望为一个比较小的常数,从 而每次进锁的程序只需要常数次测试, 如是, 转移次数期望应有上限 $O(n^2)$. 实测显示, 第 $i$ 个进锁的进程大概需要 $3i$ 次 add, get, 与前一个的差为常值 $3$.
用类似的方式, 可实现读写锁, 最简单的是加一个计数器. 在加读锁时, 按上面 方式, 加一, 减一循环, 使写锁计数器到一, 加读计数器, 减写锁计数器; 在加 写锁时, 使写锁计数器到一, 自旋至读计数器为零.

Footnotes:

¹ http://www.mcs.anl.gov/research/projects/mpi/usingmpi2/

² 我没找到非 simple 版的 MPE_Mutex_lock.

3 blogger 似乎不支持使用 Content-ID 资源的HTML邮件, 只得取消了 latex 图片展示.

部分重写密码软件

不知道是左脚站在右脚上, 还是右脚站在左脚上, 我们只知道他站在自己的脚趾上.

自上次初学 Python 写一个自己用的密码管理软件 pgman 已经有一年半了, 存在 的问题有:

• 交互程序, 不方便其它程序使用.

  比如, 有一个脚本, 初始化所有加密分区. 需要在问密码时, 切到这个程序, 调出密码到剪贴板, 再切过来, 粘贴密码.

• 密码明文存在内存里, 没有锁住, 容易被交换到磁盘, 造成泄漏.

  这个问题, 其实在目前的是没法严格解决的. 不说密码管理软件, 单使用密码 中, 流程太长, 就如浏览器等, 未必在释放内存前清空密码. 但, 在密码管理 软件中, 少一层不必要的泄漏, 总好一层.

• 访问安全问题.

  人走时, 未必记得退出此程序, 为方便, 也总是常开着. 虽然设有密码遗忘时 间. 但在别人可访问电脑时, 用 GDB 或 其它内存分析工具可完全获得所有密 码.

五一短假, 本来想用还没学会的 ErLang 练手写一个, 但 ErLang 上手不易, 以 上的问题还是无从下牙. 最后, 还是用 C 写了个.

设计上是, 使用会话加密. 会话私钥不加密地放在 U 盘里, 会话公钥放机器上. 有一个服务程序, 加载原来 pgman 的文件, 这个需要询问永久存储的私钥密码, 用会话公钥加密敏感信息, 保留结构, 存在内存里. 另有一个客户端(现在主要功 能是 BASH 脚本写的), 自动检查并挂载 U 盘, 查询信息, 解密. 服务端返回的 信息, 除状态码外, 都是加密的. RSA 块加密有一个好处是加密信息可以自然串 联起来, 而不必解密.

目前只实现读, 还没有实现写. 写需要在服务端做解密, 写用得不多, 移时再 说.

关于 RSA 加密, 测试显示, OpenSSL 的实现, 4096 位的密码操作时间是 512 位 的 120 倍, 与我所理解的 8 log(8) = 24 倍差太多了, 可能的原因是加密 的主要是短字符串, 4096 位浪费的较多.

下次等熟悉后, 考虑买一个 USB-KEY 吧, 不这样山寨了.

一湖春景

宅, 有外出玩的意向, 无咎.

三月即至, 和打油诗, 以忆未名湖春景.

垂柳拂枝怜桃花, 游人临池怜塔影. 最恨人间仲春景, 无风无雨也无晴.


2 * 2 * 3 * 277 * 6047

SWIG 磨人

每次用新工具, 我都要觉得这东西不够灵活.

SWIG 自1995 年开始, 至今也算稳定了. 只是稳定到一个不够 versatile 的状态.

SWIG的Typemap很方便, 只可惜, 找不到将多参数一起解析, 或者解析时引用前面参数的方法. 文档里倒是明确提到 numinputs 须为0或1, 像是不支持这个.

文档里找不到, 邮件列表里没人解答, 虽然还没到最后一步看实现, 估计我也放弃了. 有新资源申请依赖参数时, 手写Python扩展也比这个框架下写接口来得快.

``SWIG垃圾, 支持多语言的自然做不好.''

``框架!''

虽然, 倒是从 numpy.i 里找出一些代码, 包装数组挺方便的.

纯属测试: 中共匪然海外的都与敏感词相关么.