3
\$\begingroup\$

There's a mission-critical ring buffer in production on which I'm curious to test some optimizations. Specifically, I'm wondering how much of a performance difference struct size will make on issues like false sharing. The queue itself is single-producer, many-consumer, and the current structs that get thrown around in shared memory are 48 bytes. My theory is that, if the structs being consumed are 64 bytes, perhaps there'd be less cache misses due to false sharing.

I wrote a simplified server/consumer pair of C++ programs and a Python script to build and run them many times. What I'd like to know is: am I testing properly to explore the title-line question? That is, do these various builds even produce differently sized cache reads and writes? Perhaps more importantly, is this even a sensible thing to want to test for performance reasons?

Here's my Makefile:

CXX=g++
CFLAGS=-std=c++14 -O2 -I.
WFLAGS=-Wall -Wextra -Werror

all: server consumer

server: server.cpp data.h
    @$(CXX) $(CFLAGS) $(WFLAGS) $(ADDFLAGS) -o $@ $^

consumer: consumer.cpp data.h
    @$(CXX) $(CFLAGS) $(WFLAGS) $(ADDFLAGS) -o $@ $^

.PHONY: clean

clean:
    @rm server consumer

Here's the shared data.h:

#ifndef __DATA_H__
#define __DATA_H__

#include <cstdint>
#include <atomic>

#define QUEUE_SIZE 100

#ifdef SIZE_48B
typedef struct
{
    uint64_t price_a;
    uint64_t price_b;
    uint32_t size_a;
    uint32_t size_b;
    uint32_t ts_a;
    uint64_t ts_b;
    uint64_t padding;
}Datum;

#elif SIZE_32B

typedef struct
{
    uint64_t price_a;
    uint64_t price_b;
    uint64_t ts_b;
    uint64_t padding;
}Datum;

#elif SIZE_64B

typedef struct
{
    uint64_t price_a;
    uint64_t price_b;
    uint64_t size_a;
    uint64_t size_b;
    uint64_t ts_a;
    uint64_t ts_b;
    uint64_t padding_a;
    uint64_t padding_b;
}Datum;

#endif

typedef struct 
{
    std::atomic_uint32_t current_index;
    std::atomic_uint32_t current_turn;

    Datum data[QUEUE_SIZE];
}SharedData;

#endif //__DATA_H__

This is server.cpp:

#include <iostream>
#include <chrono>
#include <thread>
#include <csignal>
#include <cstdlib>
#include <ctime>
#include <sys/ipc.h>
#include <sys/shm.h>

#include "data.h"

bool carry_on = true;

void sig_handler(int)
{
    std::cout << std::endl;
    carry_on = false;
}

int main()
{
    signal(SIGHUP,  sig_handler);
    signal(SIGINT,  sig_handler);
    signal(SIGQUIT, sig_handler);
    signal(SIGKILL, sig_handler);
    signal(SIGTERM, sig_handler);

    srand(time(NULL));

    std::cout << "Using " << sizeof(Datum) << " byte structs" << std::endl;

    key_t key = ftok("/etc/hosts", 1);
    int shmid = shmget(key, sizeof(SharedData), 0666|IPC_CREAT);
    SharedData* share = (SharedData*) shmat(shmid, (void*) 0, 0);
    Datum* data_arr = share->data;

    share->current_turn = 0;

    while(carry_on)
    {
        for(uint32_t i = 0; i < QUEUE_SIZE; i++)
        {
#ifdef SIZE_48B
            data_arr[i].price_a = 1234567890*i;
            data_arr[i].price_b = 987654321*i;
            data_arr[i].size_a = 123*i;
            data_arr[i].size_b = 456*i;
            data_arr[i].ts_a = 789*i;
            data_arr[i].ts_b = 78901232*i;
            data_arr[i].padding = 0;
#elif SIZE_32B
            data_arr[i].price_a = 1234567890*i;
            data_arr[i].price_b = 987654321*i;
            data_arr[i].ts_b = 78901232*i;
            data_arr[i].padding = 0;
#elif SIZE_64B
            data_arr[i].price_a = 1234567890*i;
            data_arr[i].price_b = 987654321*i;
            data_arr[i].size_a = 123*i;
            data_arr[i].size_b = 456*i;
            data_arr[i].ts_a = 789*i;
            data_arr[i].ts_b = 78901232*i;
            data_arr[i].padding_a = 0;
            data_arr[i].padding_b = 0;
#endif
            share->current_index = i;

            std::chrono::microseconds duration(rand()%10);
            std::this_thread::sleep_for(duration);

            if(!carry_on) break;
        }

        if(carry_on)
        {
            share->current_index = 0;
            share->current_turn++;
            std::chrono::microseconds duration(rand()%100);
//            std::cout << share->current_turn << std::endl;
            std::this_thread::sleep_for(duration);
        }
    }

    shmdt(share);
    shmctl(shmid, IPC_RMID, NULL);

    return 0;
}

This is consumer.cpp:

#include <iostream>
#include <cstring>
#include <chrono>
#include <thread>
#include <csignal>
#include <ctime>
#include <sys/ipc.h>
#include <sys/shm.h>

#include "data.h"

using std::chrono::steady_clock;
using std::chrono::duration_cast;
using std::chrono::nanoseconds;

bool carry_on = true;

void sig_handler(int)
{
    std::cout << std::endl;
    carry_on = false;
}

int main()
{
    signal(SIGINT, sig_handler);

    srand(time(NULL));

    key_t key = ftok("/etc/hosts", 1);
    int shmid = shmget(key, sizeof(SharedData), 0666|IPC_CREAT);
    SharedData* share = (SharedData*) shmat(shmid, (void*) 0, 0);

    Datum current_datum;
    uint32_t client_index = 0;
    uint32_t client_turn = share->current_turn;
    uint32_t orig_turn = client_turn;
    uint32_t num_laps = 0;

    steady_clock::time_point start = steady_clock::now();

    while(carry_on)
    {
        if(client_turn < share->current_turn)
        {
            if(client_index < share->current_index)
            {
                num_laps++;
                client_index = share->current_index;
                client_turn  = share->current_turn;
            }
        }

        while(client_index <= share->current_index)
        {
            memcpy(&current_datum, &share->data[client_index], sizeof(Datum));
            std::chrono::microseconds duration(rand()%100);
            std::this_thread::sleep_for(duration);
            client_index++;
            if(client_index == QUEUE_SIZE)
            {
                client_turn++;
                if((client_turn - orig_turn) >= 100)
                {
                    carry_on = false;
                }
                client_index = 0;
                break;
            }
        }
    }

    steady_clock::time_point end   = steady_clock::now();
    steady_clock::duration elapsed = duration_cast<nanoseconds>(end-start);

    std::cout << elapsed.count() << " " << num_laps << std::endl;

    shmdt(share);

    return 0;
}

And finally, here's the Python script that drives it:

#!/usr/bin/python3

import subprocess
import threading
import logging
from multiprocessing import cpu_count

def run_test():
    total_runtime = []
    total_laps = []
    test_iter = 0

    while test_iter < 100:
        try: results = subprocess.run(["./consumer"], capture_output=True)
        except KeyboardInterrupt: break

        tokens = results.stdout.decode("utf-8").split(" ")
        try:
            total_runtime.append(int(tokens[0].strip()))
            total_laps.append(int(tokens[1].strip()))
        except ValueError:
            print("Error logging output. Server died?", flush=True)
            break

        test_iter += 1
#        if test_iter == 10:
#            print("  "+str(test_iter), end="", flush=True)
#        elif test_iter%10 == 0:
#            print(", "+str(test_iter), end="", flush=True)

    try:
        avg_runtime = sum(total_runtime)//len(total_runtime)
        avg_laps    = round(sum(total_laps)/len(total_laps), 2)
        one_pct_low_run = sorted(total_runtime)[1]
        one_pct_high_run = sorted(total_runtime)[-1]
        one_pct_low_lap = sorted(total_laps)[1]
        one_pct_high_lap = sorted(total_laps)[-1]
        logging.info(str(round(avg_runtime/1000/1000/1000, 6))+"s avg runtime "+
                     str(round(one_pct_low_run/1000/1000/1000, 6))+"s lo "+
                     str(round(one_pct_high_run/1000/1000/1000, 6))+"s hi")
        logging.info(str(avg_laps)+" avg laps "+str(one_pct_low_lap)+" lo "+
                     str(one_pct_high_lap)+" hi")
    except ZeroDivisionError:
        return


logging.basicConfig(level=logging.INFO, format="%(threadName)s: %(message)s")

for size in ["'-DSIZE_32B'", "'-DSIZE_48B'", "'-DSIZE_64B'"]:
    try:
        subprocess.run(["make", "clean"], capture_output=True)
        subprocess.run(["make", "all", "-j8", "ADDFLAGS="+size], check=True)
    except KeyboardInterrupt:
        quit()
    server_proc = subprocess.Popen(["./server"])

    threads = []
    for cpu in range(cpu_count()):
        t = threading.Thread(target=run_test)
        threads.append(t)
        t.start()

    for thread in threads:
        thread.join()

    server_proc.terminate()

The only further notes I can think to offer is that I've called the condition where the consumer falls behind the producer a "lap". Obviously, I want overall execution time and number of laps to be as low as possible. Unfortunately, there appears to be very little difference in the runtime or laps of the various byte sizes. =) I have to assume this is because of some poorly controlled variables and would love to have your feedback on how to tidy this test up and better optimize this data model.

\$\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.