Transaction errors

[Curtis Stanford]

Hello again! I've run into another issue with transactions during a lot of record creations. I'm still working on my akka persistence driver and they have a test you can run to check performance. All of the correctness tests pass but as soon as it gets into 10,000 events test, I get a lot of these errors:

org.cinchapi.concourse.TransactionException: Another client has made changes to data used within the current transaction, so it cannot continue. Please abort the transaction and try again.

I'm only adding records. I'm not changing any existing ones. I'm also doing everything in a transaction (reads and writes). I wouldn't think there would be any conflicts since I'm only adding new records.

The errors always seem to occur in the write portion of the code where I'm creating a new record with con.create() and populating it with stuff. No other thread changes that data once it's written.

Any thoughts appreciated.

Curtis

[Jeff Nelson]

What are the exact reads/writes you're doing? Are any of them range queries? 

JEFF NELSON

Founder & CEO | Cinchapi Inc.

www.cinchapi.com

Follow us on Twitter | Github | AngelList

--
You received this message because you are subscribed to the Google Groups "Concourse Developers" group.
To unsubscribe from this group and stop receiving emails from it, send an email to concourse-dev...@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Curtis Stanford]

This is the only write I'm doing (many times of course):

        val id = con.create()

        con.set("persistence_id", Tag.create(message.persistenceId), id)

        con.set("sequence", message.sequenceNr, id)

        val b64 = Base64.getEncoder.encodeToString(bytes)

        con.set("payload", b64, id)

I guess there is another place I do a write where I delete a range of records. So yes, I do a range query (criteria using BETWEEN operator) on sequence (Long) and delete all of the matching records (using clear(id)). I don't think the delete part is called at all in the test. It's just writing a reading a lot of stuff.

The other place I do a read is not a range query but strictly a read to find certain sequence numbers and play them back.

Curtis

[Jeff Nelson]

If your range query covers any value you are writing for a particular key in a different thread, then you'll get transaction failures.

Unlike other databases, Concourse uses just-in-time locking where transactions don't grab locks until commit time. The benefit of this is that we minimize the amount of locking that needs to happen in order to guarantee correctness. But, the side effect is that you may see more transaction fails if there is a write-read conflict. 

Let say you have two transactions, one reads a field and another writes to the same field:

t1: read x

t2:               write x

Normally, in a database, t1 would grab a lock on x and block t2 until it was finished. It Concourse, neither t1 or t2 grabs a lock until commit time. So, if t2 commits before t1, then t1 will receive an error saying another operation made a change. 

The best thing to do is wrap transactions in a try-catch block and retry if a TransactionException occurs.

JEFF NELSON

Founder & CEO | Cinchapi Inc.

www.cinchapi.com

Follow us on Twitter | Github | AngelList

[Curtis Stanford]

Yes, I did wrap them in a retry block. Plus, I'm pretty sure my ranges don't overlap the writes. I'll have to take a closer look at it. I can't see what I'm doing wrong.

[Jeff Nelson]

Can you send me the code? I can help you debug

Sent from my iPhone

[Curtis Stanford]

OK, I'll start with my method that runs code in a transaction. Not sure if I'm doing it quite right. I'm using the connection pool.

 def getWithTransaction[T](pool: ConnectionPool, body: Concourse ⇒ T): T = {

    var tries = 0

    var conn = pool.request()

    try {

      while (tries < 10) {

        try {

          conn.stage()

          val ret = body(conn)

          if (!conn.commit()) throw new TransactionException()

          return ret

        }

        catch {

          case e: TransactionException ⇒

            e.printStackTrace()

            conn.abort()

            pool.release(conn)

            conn = pool.request()

            if (tries < 10) tries += 1

            else throw e

        }

      }

      throw new TransactionException()

    }

    finally {

      if (conn != null) pool.release(conn)