Lock-free messaging stuff for communication with realtime Core Audio thread

gistfile1.m

/*!
 * Message handler fuction
 */
typedef long (*TPAudioControllerMessageHandler) (TPAudioController *THIS, long parameter1, long parameter2, long parameter3, void *ioOpaquePtr);

/*!
 * Message 
 */
typedef struct _message_t {
    TPAudioControllerMessageHandler handler;
    long parameter1;
    long parameter2;
    long parameter3;
    void *ioOpaquePtr;
    void (^responseBlock)(struct _message_t message, long response);
} message_t;

/*!
 * Message response
 */
typedef struct {
    message_t message;
    long response;
} message_response_t;




// Member variables
    TPCircularBuffer  _messageBuffer;
    TPCircularBuffer  _responseBuffer;
    NSTimer          *_responsePollTimer;
    int               _pendingResponses;


// Forward declarations

- (void)performAsynchronousMessageExchangeWithHandler:(TPAudioControllerMessageHandler)handler parameter1:(long)parameter1 parameter2:(long)parameter2 parameter3:(long)parameter3 ioOpaquePtr:(void*)ioOpaquePtr responseBlock:(void (^)(struct _message_t message, long response))responseBlock;
- (long)performSynchronousMessageExchangeWithHandler:(TPAudioControllerMessageHandler)handler parameter1:(long)parameter1 parameter2:(long)parameter2 parameter3:(long)parameter3 ioOpaquePtr:(void*)ioOpaquePtr;
- (void)pollMessageResponses;
static void processPendingMessages(TPAudioController *THIS);


// The goods



#pragma mark - Main thread-realtime thread message sending

static void processPendingMessages(TPAudioController *THIS) {
    // Only call this from the Core Audio thread, or the main thread if audio system is not yet running
    
    int32_t availableBytes;
    message_t *messages = TPCircularBufferTail(&THIS->_messageBuffer, &availableBytes);
    int messageCount = availableBytes / sizeof(message_t);
    for ( int i=0; i<messageCount; i++ ) {
        message_t* message = &messages[i];
        
        long response = message->handler(THIS, message->parameter1, message->parameter2, message->parameter3, message->ioOpaquePtr);
        
        if ( message->response_block ) {
            message_response_t message_response = { .message = *message, .response = response };
            TPCircularBufferProduceBytes(&THIS->_responseBuffer, &message_response, sizeof(message_response_t));
        }
    }
    
    TPCircularBufferConsume(&THIS->_messageBuffer, availableBytes);
}

-(void)pollMessageResponses {
    int32_t availableBytes;
    message_response_t *responses = TPCircularBufferTail(&_responseBuffer, &availableBytes);
    int responseCount = availableBytes / sizeof(message_response_t);
    for ( int i=0; i<responseCount; i++ ) {
        message_response_t *response = &responses[i];
        response->message.response_block(response->message, response->response);
        [response->message.response_block release];
        _pendingResponses--;
    }
    
    TPCircularBufferConsume(&_responseBuffer, availableBytes);
    
    if ( _pendingResponses == 0 && _responsePollTimer ) {
        [_responsePollTimer invalidate];
        _responsePollTimer = nil;
    }
}

- (void)performAsynchronousMessageExchangeWithHandler:(TPAudioControllerMessageHandler)handler parameter1:(long)parameter1 parameter2:(long)parameter2 parameter3:(long)parameter3 ioOpaquePtr:(void*)ioOpaquePtr responseBlock:(void (^)(struct _message_t message, long response))responseBlock {
    // Only perform on main thread
    if ( responseBlock ) {
        [responseBlock retain];
        _pendingResponses++;
        
        if ( _initialised && !_responsePollTimer ) {
            _responsePollTimer = [NSTimer scheduledTimerWithTimeInterval:_preferredBufferDuration target:self selector:@selector(pollMessageResponses) userInfo:nil repeats:YES];
        }
    }
    
    message_t message = (message_t) {
        .handler = handler,
        .parameter1 = parameter1,
        .parameter2 = parameter2,
        .parameter3 = parameter3,
        .ioOpaquePtr = ioOpaquePtr,
        .responseBlock = responseBlock
    };
    
    TPCircularBufferProduceBytes(&_messageBuffer, &message, sizeof(message_t));
    
    if ( !_initialised ) {
        processPendingMessages(self);
        [self pollMessageResponses];
    }
}

- (long)performSynchronousMessageExchangeWithHandler:(TPAudioControllerMessageHandler)handler parameter1:(long)parameter1 parameter2:(long)parameter2 parameter3:(long)parameter3 ioOpaquePtr:(void*)ioOpaquePtr {
    // Only perform on main thread
    __block long returned_response;
    __block BOOL finished = NO;
    
    [self performAsynchronousMessageExchangeWithHandler:handler parameter1:parameter1 parameter2:parameter2 parameter3:parameter3 ioOpaquePtr:ioOpaquePtr responseBlock:^(message_t message, long response) {
        returned_response = response;
        finished = YES;
    }];
    
    // Wait for response
    while ( !finished ) {
        [self pollMessageResponses];
        if ( finished ) break;
        [NSThread sleepForTimeInterval:_preferredBufferDuration];
    }
    
    return returned_response;
}