Core audio realtime thread non-locking sync/message passing

gistfile1.m

// Some types:

typedef struct _message_t {
    int action;
    int  kind;
    long parameter1;
    long parameter2;
    long parameter3;
    void *response_ptr;
    void (^response_block)(struct _message_t message, long response);
} message_t;

typedef struct {
    message_t message;
    long response;
} message_response_t;

enum {
    kMessageAdd,
    kMessageRemove,
    kMessageGet
};

enum {
    kKindThing,
    kKindAMoreDifferentThing
};



// Some member variables

    TPCircularBuffer  _messageBuffer;
    TPCircularBuffer  _responseBuffer;
    NSTimer            *_responsePollTimer;
    int                 _pendingResponses;


// Some forward declarations

- (void)performAsynchronousMessageExchange:(message_t)message responseBlock:(void (^)(struct _message_t message, long response))responseBlock;
- (long)performSynchronousMessageExchange:(message_t)message;
- (void)pollMessageResponses;
static void processPendingMessages(TPAudioController *THIS);

// Some implementations

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 = 0;
        
        // TODO: Perform actions here based upon the message
        
        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--;
    }
    
    if ( _pendingResponses == 0 && _responsePollTimer ) {
        [_responsePollTimer invalidate];
        _responsePollTimer = nil;
    }
}

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

- (long)performSynchronousMessageExchange:(message_t)message {
    // Only perform on main thread
    __block long returned_response;
    __block BOOL finished = NO;
    
    [self performAsynchronousMessageExchange:message responseBlock:^(message_t message, long response) {
        returned_response = response;
        finished = YES;
    }];
    
    // Wait for response
    while ( !finished ) {
        [self pollMessageResponses];
        if ( finished ) break;
        [NSThread sleepForTimeInterval:0.005];
    }
    
    return returned_response;
}