Changed RedisSubscriber wait with TaskCondition vs Timers

[etcimon]

This update provides stability and performance improvements to the RedisSubscriber:

• Uses TaskCondition notifications rather than rearming timers throughout the class.
• Permits queuing actions in case multiple tasks are communicating, the old Timer/null check method had undefined behavior.
• Fixes an error with the ConnectionPool, where connections could be re-used after they were closed. An assertion on the new event driver required this fix.

[etcimon]

Ok, this is good to go

[s-ludwig]

Fixes an error with the ConnectionPool, where connections could be re-used after they were closed.

This is not an error. ConnectionPool's design is based on the assumption that connections can be re-established. For this reason _request_reply checks if the connection is still alive before each request.

[etcimon]

Ok, I'd have to fix the driver about this assumption. Maybe a specific unit tests would help, but would it be acceptable as a feature addition to remove connections though?

[s-ludwig]

That may make sense in some situations (although it's probably more important to add support for limiting the number of connections), but it should be separately dealt with (there should be a clear vision of the use cases and implications).

[etcimon]

When you say connections can be re-established, you mean that a TCPConnection that was closed can be used to send data again without warning? Wouldn't that be prone to errors regarding different TCP options and more complex to dispose and resume at a lower level? I would have assumed using a new object would be safer

[etcimon]

Oh I see what you mean, now I realiaze I had an error in the TCPConnection.close() function that made TCPConnection.connected true for a closed connection. I'll revert the ConnectionPool changes and make the necessary changes to the driver!

[etcimon]

However, the pull is still necessary because the handling for unsubscribe/subscribe responses takes priority over publish messages, and it doesn't quite support multiplexing either, so this implementation is currently completely unsafe (old me from then was stupid ;).

[s-ludwig]

To make this more robust, the original condition should be checked here instead of relying on the return value of wait (a good guideline to follow when using condition variables in general):

synchronized (m_capMutex) {
    if (m_listening) m_captured.wait(5.seconds);
    enforce (!m_listening, "Failed to wait for Redis listener to stop.");
}

In this case, since there is nothing that m_capMutex actually protects, the synchronized block can also just be removed. But it seems like logically there should instead be a m_statusMutex that protectsm_listening and m_capture. This mutex would then be passed to both, m_captured and m_started. (always remember that the mutex is associated to the data and not to the condition variable)

A second mutex, m_connMutex may be necessary to protect the individual calls to _request_void, but that is a separate concern from the wait functionality.

[etcimon]

So, I shouldn't protect the TaskCondition with a TaskMutex? I was certain it had to be used like the pthread conditions.

Also, would you suggest using a mutex in the RedisClient to protect _request_void?

[s-ludwig]

The mutex doesn't protect the condition variable, it's more that the condition variable has a feature that it can perform the unlock-wait and trigger-lock steps atomically from the view of code that runs within a region locked by the mutex. This makes a loop like this thread-safe:

bool condition;
Mutex mutex;
Condition conditionVar;

synchronized (mutex) {
    while (!condition)
        conditionVar.wait();
    assert(condition); // only guaranteed for the mutex passed to the condition variable!
}

Using the condition variable itself would be perfectly fine without the mutex.

[etcimon]

This is ready for a review

[s-ludwig]

OT: Maybe we should add a method RedisClient.createSubscriber to create the instance. That way we get at least some freedom in the future to change what kind of storage/RC/... is used.

[s-ludwig]

Just a quick observation: Some changes in json.d ended up in this PR, they should be removed. Similarly, there are a lot of whitespace changes in redis.d, which should be removed (especially since the first thing my editor would do would be to remove the trailing WS again).

I'll try to review the actual changes later today.

[extrawurst]

i think all logInfos should be logTraces or something, so the default logging is not spammed

[extrawurst]

looks good to me, what do you think @s-ludwig ?

[s-ludwig]

Looks like there should be a scope (failure) m_lockedConn.conn.disconnect(); here, so that any parsing exception doesn't leave the connection in a bad state.

[etcimon]

This should be covered with the scope(exit) on line 927, since the pubsub_handler() is only called through the while loop under there. The only points of failure now that I think of it would be runTask and vibe.concurrency.send at line 919..., I'm not sure if these functions can throw though?

[s-ludwig]

Okay, I see. Maybe it makes sense to convert the nested functions into private member functions instead, so that the function gets a bit more manageable. But that doesn't need to be done for this PR.

[s-ludwig]

I have to admit that my review hasn't been as thorough as I had hoped, but apart from the scope thing above (btw, m_lockedConn.destroy(); should probably also be in the scope statement), it looks good to merge.

[extrawurst]

well i have to admit i am not familiar enough with the whole subscriber design in general, but the current code in trunk is simply broken. something has to be done.

[s-ludwig]

No doubt about that!