7
\$\begingroup\$

An alternative to std::vector<std::function< >> which uses tail recursion to optimize calling the functions sequentially. I'm using this on a real-time DSP thread. Because the code does single-sample processing, dispatch overhead must be minimized, since the functions (which are usually rather simple) are called 44k times/sec (Note: my target platform, iOS, does not allow JITting).

This approaches the performance of computed goto dispatch, but is extensible due to lambdas.

Half of the calling overhead on my machine (clang++ -std=c++17 -O3, YMMV):

vector<function<void()>> average time: 0.058752 seconds
func_vec<> average time:               0.025793 seconds

I suspect performance could be further improved by eliminating the virtual function dispatch. I haven't proved if allocating the functions contiguously helps (the idea is to have the lambda closures in the same cache line). Without tail recursion, performance is comparable.

Note: I'm not trying to create a drop-in replacement for std::vector<std::function<...>>.

Complete Code

#include <new>

// An array of functions which you can subsequently call.
// This allocates the functions contiguously.
// Can use tail recursion to beat vector<function<...>>
template<typename... Arguments>
class func_vec {

public:

  // Don't implement copying yet.
  func_vec(const func_vec&) = delete;
  func_vec& operator=(const func_vec&) = delete;

  func_vec() {
    _storage = new char[sizeof(end_t)];
    new (_storage) end_t;
    _size = sizeof(end_t);
    _capacity = sizeof(end_t);
  }

  ~func_vec() {
    auto h = reinterpret_cast<holder*>(_storage);
    while(h) {
      auto n = h->next();
      h->~holder();
      h = n;
    }
  }

  // Add a callable.
  template<class F>
  void push_back(F f) {

    auto sz = sizeof(callable<F>);

    // Enlarge our buffer, copying over things.
    _enlarge_by(sz);

    // Replace the end object with our callable.
    auto p = _storage+_size-sizeof(end_t);
    new (p) callable<F>(f);

    // Add a new end object.
    new (p + sz) end_t;

    _size += sz;
  }

  // Run our chain of functions.
  void execute(Arguments... args) {
    auto h = reinterpret_cast<holder*>(_storage);
    while(h) {
      h = h->call(args...);
    }
  }

  // Run with tail recursion.
  void exec_tail(Arguments... args) {
    auto h = reinterpret_cast<holder*>(_storage);
    h->call_tail(args...);
  }

private:

  struct holder {
    virtual ~holder() { }
    virtual holder* call(Arguments...) = 0;
    virtual void call_tail(Arguments...) = 0;
    virtual size_t clone_to(void* storage) = 0;
    virtual holder* next() = 0;
  };

  template<class Lambda>
  struct callable : public holder {
    Lambda lambda;
    callable(Lambda l) : lambda(l) { }
    holder* call(Arguments... args) override {
      // This call to the lambda should be inlined.
      lambda(args...);
      return this+1; // Achievement unlocked!
    }
    void call_tail(Arguments... args) override {
      lambda(args...);
      holder* next = this+1;
      next->call_tail(args...);
    }
    size_t clone_to(void* storage) override {
      new (storage) callable(lambda);
      return sizeof(callable);
    }
    holder* next() override {
      return this+1;
    }
  };

  struct end_t : public holder {
    holder* call(Arguments... args) override {
      return nullptr; // terminate iteration
    }
    void call_tail(Arguments... args) override {
      // Terminate tail recursion.
    }
    size_t clone_to(void* storage) override {
      new (storage) end_t;
      return sizeof(end_t);
    }
    holder* next() override {
      return nullptr;
    }
  };

  void _enlarge_by(size_t sz) {
    if(_size + sz >= _capacity) {

      // Reallocate and clone everything.
      _capacity = _capacity * 2 + sz;
      char* new_storage = new char[_capacity];
      size_t offset = 0;

      if(_storage) {
        auto h = reinterpret_cast<holder*>(_storage);
        while(h) {
          offset += h->clone_to(new_storage+offset);
          auto n = h->next();
          h->~holder();
          h = n;
        }
      }
      delete[] _storage;
      _storage = new_storage;
    }
  }

  char* _storage = nullptr;
  unsigned long _size = 0;
  unsigned long _capacity = 0;

};

Test Code


#include <stdio.h>
#include <stdlib.h>
#include <ctime>
#include <cassert>
#include <vector>
#include <new>
#include <functional>
#include <string>
#include <chrono>
#include "func_vec.hpp"

const int N = 1000;
const int M = 10000;

// Fill a func_vec with 10 different functions to try
// to foil the branch predictor.
template<class V>
void make_funcs(int* vars, V& v) {

  srand(0);

  for(int i=0;i<N;++i) {
    switch(rand() % 10) {
      case 0:
        v.push_back([vars] { vars[0]++; });
        break;
      case 1:
        v.push_back([vars] { vars[1]++; });
        break;
      case 2:
        v.push_back([vars] { vars[2]++; });
        break;
      case 3:
        v.push_back([vars] { vars[3]++; });
        break;
      case 4:
        v.push_back([vars] { vars[4]++; });
        break;
      case 5:
        v.push_back([vars] { vars[5]++; });
        break;
      case 6:
        v.push_back([vars] { vars[6]++; });
        break;
      case 7:
        v.push_back([vars] { vars[7]++; });
        break;
      case 8:
        v.push_back([vars] { vars[8]++; });
        break;
      case 9:
        v.push_back([vars] { vars[9]++; });
        break;

    }
  }
}

using std::vector;
using std::function;
using namespace std::chrono;

void sanity() {

    func_vec<> v;

    int x=0;

    for(int i=0;i<N;++i) {
        v.push_back([&x] { ++x; });
    }

    v.exec_tail();

    assert(x == N);

}

const int runs = 12;
const int preroll = 2; // Ignore first two runs as preroll.

void std_function_perf() {

    double t = 0;
    for(int i=0;i<runs;++i) {
        vector<function<void()>> v;

        int vars[10] = {0};
        make_funcs(vars, v);

        auto start = high_resolution_clock::now();
        for(int i=0;i<M;++i) {
           for(auto& f : v) {
               f();
           }
        }

        if(i > preroll) {
            t += duration_cast<duration<double>>(high_resolution_clock::now()-start).count();
        }

    }
    printf("vector<function<void()>> average time: %f seconds\n", t / (runs-preroll));

}

void func_vec_perf() {

    double t = 0;
    for(int i=0;i<runs;++i) {
        func_vec<> v;

        int vars[10] = {0};
        make_funcs(vars, v);

        auto start = high_resolution_clock::now();
        for(int i=0;i<M;++i) {
            v.exec_tail();
        }

        if(i > preroll) {
            t += duration_cast<duration<double>>(high_resolution_clock::now()-start).count();
        }

    }
    printf("func_vec<> average time:               %f seconds\n", t / (runs-preroll));

}

int main(int argc, char* argv[]) {

    sanity();
    std_function_perf();
    func_vec_perf();

    // I tried running each test in a separate process
    // but result were the same.
    /*
    if(argc < 2) {
        printf("usage: test_func_vec [std|func_vec]\n");
        return 0;
    }

    if(string(argv[1]) == "std") {
        std_function_perf();
    }

    if(string(argv[1]) == "func_vec") {
        func_vec_perf();
    }
    */

    printf("DONE!\n");
}

Review Goals

  1. Does it perform similarly on your machine?
  2. Have I tested performance adequately?
  3. Ideas for further performance improvement.
  4. Is there a simpler way to express this while maintaining performance?
\$\endgroup\$
2
\$\begingroup\$

Don't like this:

    if(i > preroll) {
        t += duration_cast<duration<double>>(high_resolution_clock::now()-start).count();

    }

The call to finish the clock is inside the if statement. Thus you are timing branch failure successes.

    auto end = high_resolution_clock::now();
    if(i > preroll) {
        t += duration_cast<duration<double>>(end - start).count();
    }

I think the number of iterations is way too low:

const int M = 10000;    // hard to rule out noise.
const int runs = 12;    // Maybe a couple of million calls.


const int preroll = 2;  // I suppose this helps in getting caches warm.
                        // But I would simply prefer to run it a lot times.

To make sure is no effect on memory because of one test helping the other I would run the test in both orders and average the results.

#ifndef REVERSE_TEST_ORDER

std_function_perf();
func_vec_perf();

#else

func_vec_perf();
std_function_perf();

#endif

Also I note that the average you print is the average over the number of runs. But each run executes all the numbers M times.

vector<function<void()>> average time: 0.058752 seconds
func_vec<> average time:               0.025793 seconds

So we really need to divide these numbers by another 10,000!

vector<function<void()>> average time: 0.0000058752 seconds
func_vec<> average time:               0.0000025793 seconds

Then there is 10 functions per vector. So we need to divide that by another 10.

vector<function<void()>> average time: 0.00000058752 seconds
func_vec<> average time:               0.00000025793 seconds

So .2 micro seconds against .5 micro seconds per call.

Bug

I thikn this is a bug:

// Replace the end object with our callable.
auto p = _storage+_size-sizeof(end_t);

// Need to destroy the old `end_t` so its lifetime ends before you
// can call the constructer to create an object over it.
reinterpret_cast<holder*>(p)->~holder();

// Now you can write over it.
new (p) callable<F>(f);

// Add a new end object.
new (p + sz) end_t;

ReDesign

I would separate out the resource management and business logic in func_vec. Id did this and replaced the resource management by using std::vector> and simplified the code to:

template<typename... Arguments>
class func_vec {

    private:

      struct holder {
        holder(holder* next = nullptr): next(next) {}
        virtual ~holder() { } 
        virtual holder* call(Arguments...) = 0;
        virtual void call_tail(Arguments...) = 0;
        void setNext(holder* n) {
            next = n;
        }   
        holder* next;
      };  

      template<class Lambda>
      struct callable : public holder {
        Lambda  lambda;
        callable(Lambda l, holder* next) : holder(next), lambda(l) { } 
        holder* call(Arguments... args) override {
          // This call to the lambda should be inlined.
          lambda(args...);
          return this->next;
        }   
        void call_tail(Arguments... args) override {
          lambda(args...);
          this->next->call_tail(args...);
        }   
      };  

      struct end_t : public holder {
        holder* call(Arguments... args) override {
          return nullptr; // terminate iteration
        }   
        void call_tail(Arguments... args) override {
          // Terminate tail recursion.
        }   
      };  
      std::vector<std::unique_ptr<holder>>  data;
      end_t                                 end;
    public:
      // Add a callable.
      template<class F>
      void push_back(F f) {
        std::unique_ptr<callable<F>>  next(std::make_unique<callable<F>>(f,&end));
        if (!data.empty()) {
            data.back()->setNext(next.get());
        }
        data.push_back(std::move(next));
      }

      // Run our chain of functions.
      void execute(Arguments... args) {
        holder* h = data.front().get();
        while(h != nullptr) {
            h = h->call(args...);
        }
      }

      // Run with tail recursion.
      void exec_tail(Arguments... args) {
        data.front()->call_tail(args...);
      }

};
\$\endgroup\$
  • \$\begingroup\$ Did moving the call to stop the clock change the result? I'm guessing no since that's some tiny fraction of the total work involved in the test. Each test run makes 100m calls. Also, did you see the commented out code where I tried running each test in a separate process? It made no difference in the result. Also, no there are N=1000 functions in each vector, they are just chosen at random from 10 possible functions. \$\endgroup\$ – Taylor Jul 17 at 22:23
  • \$\begingroup\$ OK. Missed the extra N. That makes me more comfortable about the number of times the function was called. N * M * (runs - preroll) = 100,000,000. \$\endgroup\$ – Martin York Jul 17 at 23:50
  • \$\begingroup\$ Did moving the call to stop the clock change the result? I did not try. I am here to review your code like a colleague would do in a big company. My suggestions would be one that I would expect to see so that once I see the results I am confident that the results are accurate. If you don't want to change fine. But then similarly I would not feel confident vouching to our boss that I though you had done a good job. \$\endgroup\$ – Martin York Jul 17 at 23:53
  • \$\begingroup\$ The results seem like approximately what I would expect. Your technique takes twice the time. But that is what I would expected as you have added an additional layer of virtual function calls. \$\endgroup\$ – Martin York Jul 17 at 23:56

Your Answer

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

Not the answer you're looking for? Browse other questions tagged or ask your own question.