생산자/소비자 다중 스레드 환경에서 공유 버퍼 최적화

Question

이벤트를 버퍼에 쓰는 단일 제작자 스레드와 버퍼에서 이벤트를받는 추가 단일 소비자 스레드가있는 프로젝트가 있습니다. 내 목표는이 작업을 단일 듀얼 코어 시스템에 최적화하여 최대 처리량을 얻는 것입니다.

현재 간단한 lock-free 링 버퍼를 사용하고 있습니다 (하나의 소비자와 하나의 제작자 스레드 만 있으므로 잠금이 가능하지 않으므로 포인터는 단일 스레드에서만 업데이트됩니다).

#define BUF_SIZE 32768 

struct buf_t { volatile int writepos; volatile void * buffer[BUF_SIZE]; 
    volatile int readpos;) }; 

void produce (buf_t *b, void * e) { 
    int next = (b->writepos+1) % BUF_SIZE; 
    while (b->readpos == next); // queue is full. wait 
    b->buffer[b->writepos] = e; b->writepos = next; 
} 

void * consume (buf_t *b) { 
    while (b->readpos == b->writepos); // nothing to consume. wait 
    int next = (b->readpos+1) % BUF_SIZE; 
    void * res = b->buffer[b->readpos]; b->readpos = next; 
    return res; 
} 

buf_t *alloc() { 
    buf_t *b = (buf_t *)malloc(sizeof(buf_t)); 
    b->writepos = 0; b->readpos = 0; return b; 
} 

그러나이 구현은 아직 충분히 빠르지 않으므로 추가로 최적화해야합니다. 나는 다른 BUF_SIZE 값을 시도하고 속도가 빨라졌다. 또한, buffer과 buffer 뒤에 writepos을 이동 했으므로 두 변수가 다른 속도로 된 다른 캐시 라인에 있는지 확인해야합니다.

내가 원하는 것은 약 400 %의 속도 향상입니다. 패딩 (padding)과 같은 것들을 사용해서 내가 어떻게 이룰 수 있었는지 아이디어가 있습니까?

Answer 1

저는 카페 응답의 첫 번째 코드 블록에서 최적화를 구현했습니다. 그들은 실제로 약간의 속도를 높이었지만 (감사합니다), 아직 충분하지 않습니다. 행 대신 열로 채워지는 캐시를 초래하는 두 번째 최적화로 인해 성능이 저하됩니다.

생산자 뒤에서 소비자가 뒤처지고 있다는 생각은 속도를 향상시키지 못했습니다.

이제 저는 300 %에 있습니다. 내가 만든

추가 변경이 일부 휘발성 writepos 및 readpos 변수에 대한 해킹했다 :

void produce (void * e) { 
    unsigned int oldpos = buffer.writepos; 
    unsigned int next = (oldpos+1) % BUF_SIZE; 
    while (next == buffer.rpos) { // rpos is not volatile 
     buffer.rpos = buffer.readpos; 
     usleep(1); 
    } 
    buffer.buffer[oldpos] = e; buffer.writepos = next; 
} 

과 소비 유사().

구조체에 대한 추가 변경 사항은 다음 새 구조체 구조체 (힙에서 대신 하나의 대답으로 제안 된 것처럼 전역 범위에서)로 이어집니다.

#define STRIDE 16 
#define STEPS 524288 

struct buf_t { 
    volatile unsigned int writepos; 
    int offset [STRIDE - 1]; 
    unsigned int wpos; 
    int offset2 [STRIDE - 1]; 
    volatile void * buffer[BUF_SIZE]; 
    int offset4 [STRIDE]; 
    volatile unsigned int readpos; 
    int offset3 [STRIDE - 1]; 
    unsigned int rpos; 
} 

300 %의 속도 향상을 얻지 못하여 누락 된 성능 제한을 밑돌았다. 당신이 더 성능을 향상하는 데 사용할 수있는 몇 가지 추가 해킹이있는 경우

, 도움도 :-)

감사를 게시 주저하지 않습니다.

Answer 2

여기서 볼 수있는 최적화 중 하나가 있습니다. consume()에서 consume()을 호출하는 스레드가 어쨌든 업데이트 할 수있는 유일한 개체이므로 b->readpos을 계속 가져올 필요가 없습니다. 또한 각 적어도 캐시 라인의 발전에 버퍼를 단계별로한다

void * consume (buf_t *b) { 
    int rp = b->readpos; 
    while (rp == b->writepos); // nothing to consume. wait 
    int next = (rp + 1) % BUF_SIZE; 
    void * res = b->buffer[rp]; b->readpos = next; 
    return res; 
} 

당신은 그렇지가 volatile을이기 때문에 당신이 명시 적으로 수행해야하므로, 컴파일러는, 멀리 모든 페치를 최적화 할 수 하나의 CPU가 캐시 라인에서 b->buffer[n]을 읽고 16 개에서 15 번을 읽고 나머지 하나가 b->buffer[n+1]을 쓰기 위해 무효화하기를 원한다면 두 개의 CPU간에 캐시 라인 핑 (ping-ponging)을 갖게됩니다. 예 :

#define STRIDE 16 
#define STEPS 2048 
#define BUF_SIZE (STRIDE * STEPS) 

#define TO_INDEX(n) (STRIDE * (((n) + 1) % STEPS) + (((n) + 1)/STEPS)) 

void produce (buf_t *b, void * e) { 
    unsigned wp = b->writepos; 
    unsigned next = (wp + 1) % BUF_SIZE; 
    while (b->readpos == next); // queue is full. wait 
    b->buffer[TO_INDEX(wp)] = e; b->writepos = next; 
} 

void * consume (buf_t *b) { 
    unsigned rp = b->readpos; 
    while (rp == b->writepos); // nothing to consume. wait 
    unsigned next = (rp + 1) % BUF_SIZE; 
    void * res = b->buffer[TO_INDEX(rp)]; b->readpos = next; 
    return res; 
} 

어쨌든 시도해 볼 가치가 있습니다. (STRIDE과 STEPS이 2의 거듭 제곱 인 한, TO_INDEX()의 무서운 디비전 및 모듈러스는 피연산자가 unsigned 인 경우에만 shift 및 bitwise-and로 최적화 할 수 있으므로 이에 따라 해당 유형을 변경하는 것이 좋습니다.).

Answer 3

두 개 이상의 프로세서 또는 코어가있는 컴퓨터를 사용한다고 가정합니다. 그렇지 않다면 당신의 바쁜 기다림이 사물을 다칠 것입니다. 어쨌든 당신이 충분히 잠을 자지 못하도록 결정하고 동적 우선 순위를 낮추고 다른 프로그램이 실행 중이라고 결정한 운영 체제에서 실행 중이면 어쨌든 다칠 수 있습니다.

버퍼 용량이 얼마나되는지에 대한 데이터를 수집해야합니다. 캐시가 너무 커지면 캐시가 손상 될 수 있습니다.

힙에서 할당하지 않고 전역 배열을 사용하는 경우 버퍼에 대한 포인터가 포인터 리터럴이되고 두 스레드가 캐시에서 같은 위치의 포인터 값을 읽을 필요가 없습니다. 코드에 빠져 들었다.

처리량이 중요한 경우 (지연 시간을 희생하여) 캐시가 실제로 큰 롤을 돌리고 있다면 소비자가 생산자를 뒤처지게하여 해당 캐시에서 읽고 쓰려고하지 않는 것을 고려할 수 있습니다. 버퍼 내의 같은 장소.

인터페이스를 소비자 함수로 변경하여 캐쉬 크기 (또는 여러 개의) 크기의 청크 (위의 제작자 제안보다 뒤처진 소비자에게 좋음)를 소비 할 수 있습니다. 부분 캐시 라인 청크. 소비 캐시를 정렬 된 상태로 유지하십시오. 사용 가능한 데이터를 뱀으로 생각하면 머리와 꼬리가 정렬되지 않을 수 있습니다. 소비 할 다른 데이터가 없을 때만 정렬되지 않은 꼬리를 사용해야합니다. 통화에서 다른 데이터를 소비하여 소비하는 경우 다음 통화를 위해 꼬리를 남겨 두어야합니다.

이외에 caf에서 언급 한 바에 따르면이 코드 외부에서 일어나는 일이 무엇이든간에 더 큰 역할을해야한다고 생각합니다.