Несовместимые моменты времени при передаче данных между двумя потоками

У меня есть кусок кода, который я использую для тестирования различных контейнеров (например, deque и кольцевой буфер) при передаче данных от производителя (поток 1) к потребителю (поток 2). Данные представлены структурой с парой временных меток. Первая временная метка берется перед отправкой производителю, а вторая — когда данные извлекаются потребителем.
Контейнер защищен прядильным устройством.

Машина работает на Redhat 5.5 с ядром 2.6.18 (старое!), Это 4-ядерная система с отключенной гиперпоточностью. gcc 4.7 с флагом -std = c ++ 11 использовался во всех тестах.

Producer получает блокировку, ставит метки времени в данные и помещает их в очередь, разблокирует и спит в цикле занятости в течение 2 микросекунд (единственный надежный способ, который я нашел, чтобы спать только для 2 микросхем в этой системе).

Потребитель блокирует, извлекает данные, ставит метки времени и генерирует некоторую статистику (средняя задержка и стандартное отклонение). Статистика печатается каждые 5 секунд (M — среднее значение, M2 — стандартное отклонение) и сбрасывается. Я использовал gettimeofday () для получения меток времени, что означает, что среднее число задержек можно рассматривать как процент задержек, превышающих 1 микросекунду.

Большую часть времени вывод выглядит так:

    CNT=2500000 M=0.00935 M2=0.910238
CNT=2500000 M=0.0204112 M2=1.57601
CNT=2500000 M=0.0045016 M2=0.372065

но иногда (вероятно, 1 испытание из 20), как это:

    CNT=2500000 M=0.523413 M2=4.83898
CNT=2500000 M=0.558525 M2=4.98872
CNT=2500000 M=0.581157 M2=5.05889

(обратите внимание, что среднее число намного хуже, чем в первом случае, и оно никогда не восстанавливается при запуске программы).

Буду признателен за мысли о том, почему это может произойти. Благодарю.

#include <iostream>
#include <string.h>
#include <stdexcept>
#include <sys/time.h>
#include <deque>
#include <thread>
#include <cstdint>
#include <cmath>
#include <unistd.h>
#include <xmmintrin.h> // _mm_pause()

int64_t timestamp() {
struct timeval tv;
gettimeofday(&tv, 0);
return 1000000L * tv.tv_sec + tv.tv_usec;
}

//running mean and a second moment
struct StatsM2 {
StatsM2() {}
double m = 0;
double m2 = 0;
long count = 0;
inline void update(long x, long c) {
count = c;
double delta = x - m;
m += delta / count;
m2 += delta * (x - m);
}
inline void reset() {
m = m2 = 0;
count = 0;
}
inline double getM2() { // running second moment
return (count > 1) ? m2 / (count - 1) : 0.;
}
inline double getDeviation() {
return std::sqrt(getM2() );
}
inline double getM() { // running mean
return m;
}
};

// pause for usec microseconds using busy loop
int64_t busyloop_microsec_sleep(unsigned long usec) {
int64_t t, tend;
tend = t = timestamp();
tend += usec;
while (t < tend) {
t = timestamp();
}
return t;
}

struct Data {
Data() : time_produced(timestamp() ) {}
int64_t time_produced;
int64_t time_consumed;
};

int64_t sleep_interval = 2;
StatsM2 statsm2;
std::deque<Data> queue;
bool producer_running = true;
bool consumer_running = true;
pthread_spinlock_t spin;

void producer() {
producer_running = true;
while(producer_running) {
pthread_spin_lock(&spin);
queue.push_back(Data() );
pthread_spin_unlock(&spin);
busyloop_microsec_sleep(sleep_interval);
}
}

void consumer() {
int64_t count = 0;
int64_t print_at = 1000000/sleep_interval * 5;
Data data;
consumer_running = true;
while (consumer_running) {
pthread_spin_lock(&spin);
if (queue.empty() ) {
pthread_spin_unlock(&spin);
// _mm_pause();
continue;
}
data = queue.front();
queue.pop_front();
pthread_spin_unlock(&spin);
++count;
data.time_consumed = timestamp();
statsm2.update(data.time_consumed - data.time_produced, count);
if (count >= print_at) {
std::cerr << "CNT=" << count << " M=" << statsm2.getM() << " M2=" << statsm2.getDeviation() << "\n";
statsm2.reset();
count = 0;
}
}
}

int main(void) {
if (pthread_spin_init(&spin, PTHREAD_PROCESS_PRIVATE) < 0)
exit(2);
std::thread consumer_thread(consumer);
std::thread producer_thread(producer);
sleep(40);
consumer_running = false;
producer_running = false;
consumer_thread.join();
producer_thread.join();
return 0;
}