Sum a list of string numbers

Question

I have a list of 32-bit signed numbers in range [-10e5, 10e5] in string form read from stdin, and I would like to compute and print their sum in the fastest way possible.

The bottleneck is in the fast_atoi function (checked using callgrind)

I thinking of two ways for improvement : cache misses seems to be an issue, and SIMD (ideally, only SSE and SSE2) instructions could come to help, however i don't really know how to use them...

The input is in this form (signed 32-bit integers between -10e5 and 10e5, separated by '\n') :

4
10
5
3
1

And the program should output :

19

Right now, my code is :

#include <cstdint>
#include <cstdlib>
#include <cstdio>
#include <cstring>
#include <unistd.h>
#include <limits.h>
#include <sys/stat.h>

#define likely(x)      __builtin_expect(!!(x), 1)
#define unlikely(x)    __builtin_expect(!!(x), 0)

char* data alignas(16) = nullptr;

inline void read_data()
{
    struct stat sb;
    int rc = fstat( STDIN_FILENO, &sb );
    data = (char*)malloc(sb.st_size + 1 );
    size_t totalRead = 0UL;
    while (totalRead  < sb.st_size)
    {
        ssize_t bytesRead = read(STDIN_FILENO, data + totalRead, sb.st_size - totalRead);
        if ( bytesRead <= 0 )
        {
            break;
        }
        totalRead += bytesRead;
    }
}

inline int fast_atoi(char ** str )
{
    int val = 0;
    char neg = 1;
    if (**str == '-')
    {
        neg = -1;
        (*str)++;
    }
    while(unlikely(**str != '\n'))
    {
        val = val*10 + (*(*str)++ - '0');
    }
    ++*str;
    return val*neg;
}
int32_t process()
{
    size_t count = fast_atoi(&data);
    int32_t sum = 0;
    for (unsigned ii = 0; unlikely(ii < count); ii++ )
    {
        sum += fast_atoi(&data);
    }
    return sum;
}
inline void write_ans(int32_t sum)
{
    printf("%d", sum);
}
int main()
{
    read_data();
    int32_t val = process();
    write_ans(val);
}

Callgrind report for this code : https://pastebin.com/Pnd0FzZb Assembly output : https://godbolt.org/g/Ng3sVM

Answer 1

I see a few things that may help you improve your code.

Add some error checking

Many of the calls used in the code, including fstat and malloc can fail. It's generally better to check for and handle such errors instead of ignoring them. Even though your main concern is performance, these calls are only made once and so will add almost nothing to the runtime.

Here's an alternative main showing how that might look:

int main() {
    struct stat sb;
    char *data;
    if (fstat(STDIN_FILENO, &sb) != 0 || (data = (char *)mmap(0, sb.st_size, PROT_READ, MAP_SHARED, STDIN_FILENO, 0)) == nullptr) {
        perror("Error reading stdin");
        return 1;
    } 
    std::cout << sum(data, sb.st_size) << '\n';
}

Avoid the use of global variables

The data variable is used within read_data and the poorly named process(). It's generally better to explicitly pass variables your function will need or declare them within the appropriately smallest possible scope rather than using the vague implicit linkage of a global variable.

Avoid premature optimization

On my machine, omitting the use of the unlikely macro and the alignas(16) made no difference in the resulting speed of the program. Generally, I only resort to such things if there is a measurable effect and performance is more important than portability. Perhaps you've already measured and made this determination and perhaps the results are different on your machine.

Consider an alternative algorithm

An alternative algorithm, which yields times very similar to the ones with your original, might be employed here. In particular, we could process the file from the end to the beginning instead of the usual direction. This allows us to efficiently convert each number, only adding the converted number when encountering a newline. Note that this also essentially throws away the first number (which is the count of numbers to follow) which is the correct behavior in this case, since that number is not intended to be included in the sum. Here's one version of the sum routine for that algorithm:

int sum(const char *test, std::size_t len) {
    int result{0};
    int n{0};
    int mult{1};
    // start from the back and add digits
    for (const char *ptr = test+len-1; ptr >= test; --ptr) {
        switch(*ptr) {
            case '\n':
                result += n;
                n=0;
                mult=1;
                break;
            case '-':
                n = -n;
                break;
            case '0':
            case '1':
            case '2':
            case '3':
            case '4':
            case '5':
            case '6':
            case '7':
            case '8':
            case '9':
                n += mult * (*ptr - '0');
                mult *= 10;
                break;
            default:
                std::cerr << "Malformed input\n";
                exit(1);
        }
    }
    return result;
}

Answer 2

Likely or unlikely?

I didn't run your code, but something didn't make sense to me when I was reading it. You use an unlikely macro in several of your loops. But it seems to me that these should be likely instead of unlikely. For example:

while(unlikely(**str != '\n'))

Here, I would expect the condition to be likely, because there are more digit characters than newlines. So I think you are actually sabotaging your code by mispredicting your branches.

However, according to this stackoverflow question, branch prediction hints are largely ignored on x86 targets, so your likely/unlikely macros will have very little effect unless you are planning to run on a non-x86 target.

Answer 3

Personally, I prefer dereferencing the pointer to pointer once (before the loop), and rewriting it at the and of the loop. When inlined, the compiler will happily optimize it away. It is also clearer to the human eye.

static int oneint(char**pp){
int ch,val,sign;
char *str;

sign = 0;
val= 0;
for(str = *pp; *str; str++) {
        ch= *str;
        switch(ch) {
        case '\n':
                str++;
                goto done;
                break;
        case '-':
                sign=1; 
                break;
        default:
                // if(ch <'0' || ch> '9') continue; /* shouldneverhappen */
                val = val*10 + ch- '0';
                break;
        }}
done:
val = (sign) ? -val : val;
*pp = str;
return val;
}

---

Newer version with two simplified loops:

static int oneint2(char**pp){
int val;
char *str;

val= 0;
str = *pp;

if(*str == '-') for(str++ ; *str; str++) {
        int ch;
        ch= *str;
        if(ch == '\n'){ str++; goto done;}
        val = val*10 - (ch - '0');
        }
else for(       ; *str; str++) {
        int ch;
        ch= *str;
        if(ch == '\n'){ str++; goto done;}
        val = val*10 + ch - '0';
        }
done:
*pp = str;
return val;
}

Answer 4

Hmm interesting problem, I think I have an solution that might be faster (but I'm not able to test at the moment).

You said that atoi was the bottle neck and I find this reasonable. So lets work around atoi. Recall that base 10 numbers are represented as \$x=\sum_{i=0}^\infty a_i10^i\$ so a sum of numbers \$X=x_0,...,x_n=\sum_{j=0}^nx_j=\sum_{i=0}^\infty (\sum_{j=0}^n a_{ij})10^i\$.

Which means that if we maintain an array of int for the sum of each \$a_i\$ we can reduce the cost to effectively one summed atoi.

Something like this should work:

// Assumes only digits, '-' or '\n' present in the buffer.
// Numbers are separated by newline and have at most 8 digits 
// excluding sign.
int sumOfBuffer(const std::vector<char>& buffer){
    int max_digits = 8;
    int digit = 0;
    int baseCurrent[max_digits] = // init to zero.
    int baseSums[max_digits] = //init to zero.

    auto it = buffer.rbegin();

    while(it != buffer.rend() && '\n' == *it){
        it++;
    }

    while(it != buffer.rend()){
        if(*it == '-'){
            for(int i = 0; i < digit; ++i){
                // One can use SIMD for these loops, just round digit
                // up to closest multiple of SIMD unit size.
                // Or simply do all digits, compiler might even do
                // it for you. 
                baseSums[i] -= baseCurrent[i];
                baseCurrent[i] = 0;
            }
            // Skip past the next newline
            it++;
            while(it != buffer.rend() && '\n' != *it){
                it++;
            }
            // IIRC all past the end iterators must compare equal so 
            // this should be safe even if the above terminated due to 
            // going out of bounds. But check the standard first, I'm not
            // 100% on this. Otherwise add check for rend()
            it++;
            digit = 0;
        }
        else if(*it == '\n'){
            for(int i = 0; i < digit; ++i){
                baseSums[i] += baseCurrent[i];
                baseCurrent[i] = 0;
            }
            it++;
            digit = 0;
        }else{
            baseCurrent[digit++] = *(it++) - '0';
        }
    }

    int ans = 0;
    for(int i = max_digits-1; i >= 0; --i){
        ans = ans * 10 + baseSums[i];
    }
    return ans;
}

Please excuse the probably buggy code, it's late and I need to hit the hay. I'll fix it up tomorrow.

Answer 5

Solved by directly mmaping stdin to a buffer (0.06s -> 0.05s)

inline void read_data()
{
    struct stat sb;
    int rc = ::fstat( STDIN_FILENO, &sb );
    (void)rc; // silencer le warning
    data = (char*)mmap (0, sb.st_size, PROT_READ, MAP_SHARED, STDIN_FILENO, 0);
    upBound = data + sb.st_size;
}