Race condition when executing multiple queries scheduled synchronously.

[pulyaevskiy]

Hi,

Thanks for your great work on this library!

This PR is intended to fix an issue when multiple queries being scheduled synchronously and executed in parallel.

I've added a test case, but I'm not sure if it'll work with the rest of test suite, most likely not. The setup is quite complicated. If you can provide some assistance with that it would be great!

[achilleasa]

Hello! Thanks for the PR.

All driver tests use a mock server which replays cassandra packets captured by wireshark.

You can copy this block to your race condition test.

Something like (no setup/teardown required):

server.setReplayList(["select_v2.dump","select_v2.dump"]);
client = new cql.Client.fromHostList([ "${SERVER_HOST}:${SERVER_PORT}"]
        , poolConfig : new cql.PoolConfiguration(autoDiscoverNodes : false)
);

expect(Future.Wait([
client.execute(new cql.Query("SELECT * from test.type_test")),
client.execute(new cql.Query("SELECT * from test.type_test"))
]), completes)

Let me know if you have any issues with it. Once the test works with the rest of the test suite I will merge it .

[pulyaevskiy]

Thanks!

I updated the test, but there seems to be another race condition in the mock server implementation so it fails now (with real Cassandra server it passes). The error I get:

➜  dart_cassandra_cql git:(version-update) ✗ dart --checked test/lib/race_condition_test.dart
unittest-suite-wait-for-done
[INFO]  [2016-02-17 09:39:39.278677]    [MockLogger]:   Binding MockServer to 127.0.0.1:32000
[INFO]  [2016-02-17 09:39:39.291658]    [MockLogger]:   [127.0.0.1:32000] Listening for incoming connections
[INFO]  [2016-02-17 09:39:39.310568]    [dart_cassandra_cql.ConnectionPool]:    Initializing pool connections
[INFO]  [2016-02-17 09:39:39.312031]    [dart_cassandra_cql.Connection]:    [127.0.0.1:32000-0] Trying to connect to 127.0.0.1:32000 [attempt 1/10]
[INFO]  [2016-02-17 09:39:39.324094]    [MockLogger]:   Client [127.0.0.1:58628] connected
[FINE]  [2016-02-17 09:39:39.336428]    [dart_cassandra_cql.Connection]:    [127.0.0.1:32000-0] [stream #0] Sending message of type STARTUP (0x1) Instance of 'StartupMessage'
[FINE]  [2016-02-17 09:39:39.366510]    [dart_cassandra_cql.Connection]:    [127.0.0.1:32000-0] [stream #0] Received message of type READY (0x2) Instance of 'ReadyMessage'
[INFO]  [2016-02-17 09:39:39.367229]    [dart_cassandra_cql.Connection]:    [127.0.0.1:32000-0] Connected to 127.0.0.1:32000 [attempt 1/10]
[INFO]  [2016-02-17 09:39:39.368231]    [dart_cassandra_cql.ConnectionPool]:    Listening for server events on [127.0.0.1:32000-0]
[FINE]  [2016-02-17 09:39:39.369389]    [dart_cassandra_cql.Connection]:    [127.0.0.1:32000-0] [stream #1] Sending message of type REGISTER (0xb) Instance of 'RegisterMessage'
[FINE]  [2016-02-17 09:39:39.370766]    [dart_cassandra_cql.Connection]:    [127.0.0.1:32000-0] [stream #2] Sending message of type QUERY (0x7) Instance of 'QueryMessage'
[FINE]  [2016-02-17 09:39:39.372129]    [dart_cassandra_cql.Connection]:    [127.0.0.1:32000-0] [stream #3] Sending message of type QUERY (0x7) Instance of 'QueryMessage'
FAIL: Race Conditions: it can handle execution of multiple queries scheduled synchronously
  Caught Bad state: StreamSink is bound to a stream
  dart:io/io_sink.dart 178                              _StreamSinkImpl.flush
  dart:io-patch/socket_patch.dart 1551                  _Socket.flush
  test/lib/mocks/mocks.dart 227:35                      MockServer.replayFile.onReplay
  test/lib/mocks/mocks.dart 235:19                      MockServer.replayFile
  test/lib/mocks/mocks.dart 279:7                       MockServer._handleClientFrame
  test/lib/mocks/mocks.dart 298:28                      MockServer._handleConnection.<fn>
...

Wonder if you have any thoughts on this? Otherwise I'll try to fix this later this week.

[achilleasa]

I have pushed a fix for the race condition in MockServer to my master branch. If you rebase and add the following teardown block to your test all tests should pass.

teardown((){
   return server.shutdown();
});

[pulyaevskiy]

Thanks!

I updated my branch, though commit history got messed up after all rebases and merges.

The test suite is green now.
I moved the test case to client_test.dart in the execute: group, since it seems more relevant to what we're actually testing.

[achilleasa]

Thanks again for your PR.

I have published version 0.1.5 that includes your fix.