The lack of asyncIncr() with default -> could lead atomicity losses?

Hi Experts!

I hope I'm not missed something in the Java API (I know it depends heavily on spymemcached), but I would like to ask, the above question.

This is the relevant part in my test update() method:

 OperationFuture<Long> future = client.asyncIncr(key, pojo.getValue());
 
 latch.increment();
 
 future.addListener(new OperationCompletionListener() {
 
 @Override
 public void onComplete(final OperationFuture<?> future) throws Exception {
 
  // first time creation
  if ((Long) future.get() == -1) {
    client.set(key, Long.toString(pojo.getValue()));
 
  } else {
 
    if (!future.isDone()) {
      System.out.println("@@@ huhhh needs a second try! " + key);
    }
  }
 
  latch.decrement();
 
  logStat(name, startTime);
 }
});

To do this job in one API call could be -> more safer and faster under heavy situations.

Thank You,
Loolek

2 Answers

« Back to question.

Hi Daschl!

Thank you very much, you are very kind =)

I will test your fix!

My current job is perftesting (on the same cluster) the Couchbase Server agains Apache Cassandra.

We are needing/testing only one feature -> but that heavily depends on asncIncr()!

Cheers,
Loolek

Hey Daschl!

I have to implement a (we call it) Couchbase Future Scheduler impl -> because I'm getting too fast after adding your PATCH => I'v got TOO MANY FUTURE TIMEOUT!

1.) let's implement it

Configured to my laptop -> I had to add two lines of code ;)

@Override
public void update(final MetricsPojo pojo, final Generator generator, final SimpleLatch latch) throws IllegalStateException {

// let the future come reality :PPP
while (latch.get() > 5000) {
Tool.sleep(250);
}

Cheers,
Loolek

« Back to question.

Hi loolek,

you are right - and the protocol semantics would even allow it. In one of the next minor releases with can expose it as part of the interface.

I created a ticket that you can track: http://www.couchbase.com/issues/browse/SPY-152

Note that the sync methods have it exposed.

If you want to go for an adventure and can't wait that it gets exposed, check this out: https://gist.github.com/daschl/8600479

Hi Daschl!

I'v checked your git right now and -> that would.be cool =)

Thank you!

ps:

/**
* 1:1 copied from MemcachedClient because it has private access. gotta fix this in a future release.
*/

LOL

Cheers,
Loolek

Hi Daschl!

I'v just added your MyCouchbaseClient snippet -> than just using this new method call =)

OperationFuture future = client.asyncIncr(key, pojo.getValue(), pojo.getValue(), EXP_ONE_YEAR);

And everything works fine -> Thank You!

Your snippet was so funky =)

I had to change (I mean delete) only a few lines ;)

@Override
public void connect() throws IOException, ConfigurationException, URISyntaxException {
(snip)
 
	client = new MyCouchbaseClient(Arrays.asList(new URI("http://127.0.0.1:8091/pools")), "metrics", "");
 
	LOGGER.info(Main.PREFIX + "connected to cluster done bucket = " + bucket);
}
 
 
@Override
public void update(final MetricsPojo pojo, final Generator generator, final SimpleLatch latch) throws IllegalStateException  {
(snip)
 
	// OKAY HERE WE GO ////////////////////////////////////////////////////
 
 	OperationFuture<Long> future = client.asyncIncr(key, pojo.getValue(), pojo.getValue(), EXP_ONE_YEAR);
 
 	latch.increment();
 
 	future.addListener(new OperationCompletionListener() {
 
 		@Override
		public void onComplete(final OperationFuture<?> future) throws Exception {
 
// 			System.out.println("# listener completed f:" + future.isDone() + " k:'" + key + "' (" + latch.get() + ")"); 	
 
			try {	
				if (!future.isDone()) {
					LOGGER.error(Main.PREFIX + "huhhh needs a second try! " + key);
				}
 
				if ((Long) future.get() == -1) {
					LOGGER.error(Main.PREFIX + "first attampt failed future.get = -1");
				}
 
			} catch (ExecutionException e) {
				LOGGER.error(Main.PREFIX + "write future timeout! " + key, latch.get());
			}
 
			latch.decrement();
 
			// now we will count and stat the FAILED futures too! 
			logStat(name, startTime);
		}
	});
 }

ps:

I can not wait until monday -> to get the new perftesing result from the cluster -> against the last run"s results!

Cheers,
Loolek

>> I can not wait until monday -> to get the new perftesting results from the cluster...

Now I'v just used my fiveyears old laptop for a single perftest -> bu the results getting a little bit better :D

02:43:47.823 [gen-thread-4] INFO  c.a.perf.UpdateMetricsThread - # thread done [gen-thread-4]
02:43:48.374 [main] INFO  com.loolek.perf.Runner - # ----------------------------------------------------------------------
02:43:48.374 [main] INFO  com.loolek.perf.Runner - # waiting to finish all write future = 1033
02:43:48.574 [main] INFO  com.loolek.perf.Runner - # all timers stoped done
02:43:48.575 [main] INFO  com.loolek.perf.Reporter - # ======================================================================
02:43:48.575 [main] INFO  com.loolek.perf.Reporter - # THE STATISTICS
02:43:48.575 [main] INFO  com.loolek.perf.Reporter - # ======================================================================
02:43:48.576 [main] INFO  com.loolek.perf.Reporter - # timer nanos [ gen-thread-1 min: 508054316 max: 1322869416 ]
02:43:48.576 [main] INFO  com.loolek.perf.Reporter - # timer nanos [ gen-thread-2 min: 506638111 max: 1324653894 ]
02:43:48.576 [main] INFO  com.loolek.perf.Reporter - # timer nanos [ gen-thread-3 min: 506468732 max: 1323067026 ]
02:43:48.576 [main] INFO  com.loolek.perf.Reporter - # timer nanos [ gen-thread-4 min: 506325443 max: 1327109903 ]
02:43:48.576 [main] INFO  com.loolek.perf.Reporter - # all readed records = 0
02:43:48.576 [main] INFO  com.loolek.perf.Reporter - # all long to string convert = 383982
02:43:48.576 [main] INFO  com.loolek.perf.Reporter - # all updated records = 499996
02:43:48.576 [main] INFO  com.loolek.perf.Reporter - # all update time summa = 90.229514971 sec
02:43:48.576 [main] INFO  com.loolek.perf.Reporter - # one update time min/max/avg = 0,506325 / 1,327110 /<b> 0,000180 sec</b>
02:43:48.577 [main] INFO  com.loolek.perf.Reporter - # update per second min/max/avg = 1 / 2 /</b> *** 5541 *** </b>
02:43:48.577 [main] INFO  com.loolek.perf.Reporter - # ======================================================================
02:43:48.578 [main] INFO  c.a.perf.db.DatabaseImplCouchbase - # call cluster shutdown
02:43:48.586 [Memcached IO over {MemcachedConnection to 127.0.0.1/127.0.0.1:11210} - SHUTTING DOWN (telling client)] INFO  c.c.client.CouchbaseConnection - Shut down Couchbase clie
nt
02:43:48.701 [Couchbase View Thread] INFO  com.couchbase.client.ViewConnection - I/O reactor terminated
02:43:48.701 [main] INFO  c.a.perf.db.DatabaseImplCouchbase - # shutdown was successful = true
02:43:48.701 [main] INFO  com.loolek.perf.Main - # normal exit :)
02:43:48.702 [shutdown-thread] INFO  com.loolek.perf.Runner - # shutdown hook in progress...

ps: Until the new Java Driver release -> I will miss the builder options :(

cfb.setTimeoutExceptionThreshold(256);

cfb.setOpTimeout(5000); // set operand timeout (default 2500)

cfb.setObsTimeout(10000); // set observer timeout (default 5000)

Cheers,
Loolek

>> I can not wait until monday -> to get the new perftesting results from the cluster...

Now I'v just used my fiveyears old laptop for a single perftest -> bu the results getting a little bit better :D

02:43:47.823 [gen-thread-4] INFO  c.a.perf.UpdateMetricsThread - # thread done [gen-thread-4]
02:43:48.374 [main] INFO  com.loolek.perf.Runner - # ----------------------------------------------------------------------
02:43:48.374 [main] INFO  com.loolek.perf.Runner - # waiting to finish all write future = 1033
02:43:48.574 [main] INFO  com.loolek.perf.Runner - # all timers stoped done
02:43:48.575 [main] INFO  com.loolek.perf.Reporter - # ======================================================================
02:43:48.575 [main] INFO  com.loolek.perf.Reporter - # THE STATISTICS
02:43:48.575 [main] INFO  com.loolek.perf.Reporter - # ======================================================================
02:43:48.576 [main] INFO  com.loolek.perf.Reporter - # timer nanos [ gen-thread-1 min: 508054316 max: 1322869416 ]
02:43:48.576 [main] INFO  com.loolek.perf.Reporter - # timer nanos [ gen-thread-2 min: 506638111 max: 1324653894 ]
02:43:48.576 [main] INFO  com.loolek.perf.Reporter - # timer nanos [ gen-thread-3 min: 506468732 max: 1323067026 ]
02:43:48.576 [main] INFO  com.loolek.perf.Reporter - # timer nanos [ gen-thread-4 min: 506325443 max: 1327109903 ]
02:43:48.576 [main] INFO  com.loolek.perf.Reporter - # all readed records = 0
02:43:48.576 [main] INFO  com.loolek.perf.Reporter - # all long to string convert = 383982
02:43:48.576 [main] INFO  com.loolek.perf.Reporter - # all updated records = 499996
02:43:48.576 [main] INFO  com.loolek.perf.Reporter - # all update time summa = 90.229514971 sec
02:43:48.576 [main] INFO  com.loolek.perf.Reporter - # one update time min/max/avg = 0,506325 / 1,327110 / 0,000180 sec
02:43:48.577 [main] INFO  com.loolek.perf.Reporter - # update per second min/max/avg = 1 / 2 / *** 5541 ***
02:43:48.577 [main] INFO  com.loolek.perf.Reporter - # ======================================================================
02:43:48.578 [main] INFO  c.a.perf.db.DatabaseImplCouchbase - # call cluster shutdown
02:43:48.586 [Memcached IO over {MemcachedConnection to 127.0.0.1/127.0.0.1:11210} - SHUTTING DOWN (telling client)] INFO  c.c.client.CouchbaseConnection - Shut down Couchbase clie
nt
02:43:48.701 [Couchbase View Thread] INFO  com.couchbase.client.ViewConnection - I/O reactor terminated
02:43:48.701 [main] INFO  c.a.perf.db.DatabaseImplCouchbase - # shutdown was successful = true
02:43:48.701 [main] INFO  com.loolek.perf.Main - # normal exit :)
02:43:48.702 [shutdown-thread] INFO  com.loolek.perf.Runner - # shutdown hook in progress...

ps:

Until the new Java Driver release -> I think, I will miss these builder options :(

		cfb.setTimeoutExceptionThreshold(256);
 
		cfb.setOpTimeout(5000);  // set operand timeout (default 2500)
 
		cfb.setObsTimeout(10000);  // set observer timeout (default 5000)

Cheers,
Loolek